759 lines
27 KiB
Haskell
759 lines
27 KiB
Haskell
|
|
module Nuthatch.App
|
||
|
|
( Env (..)
|
||
|
|
, newEnv
|
||
|
|
, application
|
||
|
|
, runRecovery
|
||
|
|
) where
|
||
|
|
|
||
|
|
import Control.Concurrent (threadDelay)
|
||
|
|
import Control.Exception (SomeException, try)
|
||
|
|
import Control.Monad (forever, void, when)
|
||
|
|
import Crypto.Hash (Digest, SHA256, hashlazy)
|
||
|
|
import qualified Data.Aeson as A
|
||
|
|
import qualified Data.ByteArray as BA
|
||
|
|
import qualified Data.ByteString as BS
|
||
|
|
import qualified Data.ByteString.Base16 as B16
|
||
|
|
import qualified Data.ByteString.Char8 as BS8
|
||
|
|
import qualified Data.ByteString.Lazy as BL
|
||
|
|
import Data.Maybe (fromMaybe, listToMaybe)
|
||
|
|
import Data.Text (Text)
|
||
|
|
import qualified Data.Text as T
|
||
|
|
import qualified Data.Text.Encoding as TE
|
||
|
|
import Data.UUID (toText)
|
||
|
|
import Data.UUID.V4 (nextRandom)
|
||
|
|
import Database.PostgreSQL.Simple (Connection, Only (..))
|
||
|
|
import Network.HTTP.Types
|
||
|
|
( Header
|
||
|
|
, HeaderName
|
||
|
|
, Status (..)
|
||
|
|
, hContentDisposition
|
||
|
|
, hContentType
|
||
|
|
, hLocation
|
||
|
|
, status200
|
||
|
|
, status302
|
||
|
|
, status400
|
||
|
|
, status403
|
||
|
|
, status404
|
||
|
|
, status413
|
||
|
|
, status429
|
||
|
|
, status500
|
||
|
|
, status503
|
||
|
|
)
|
||
|
|
import Network.Wai
|
||
|
|
( Application
|
||
|
|
, Request (..)
|
||
|
|
, Response
|
||
|
|
, ResponseReceived
|
||
|
|
, responseLBS
|
||
|
|
)
|
||
|
|
import Network.Wai.Parse (FileInfo (..), lbsBackEnd, parseRequestBody)
|
||
|
|
import Nuthatch.Audit (audit)
|
||
|
|
import Nuthatch.Config (Config (..))
|
||
|
|
import Nuthatch.Db (Db, ensurePersonalPool, withTransaction)
|
||
|
|
import qualified Nuthatch.Db as Db
|
||
|
|
import Nuthatch.Paths
|
||
|
|
import Nuthatch.Protocol
|
||
|
|
import Nuthatch.Redis (Redis, cacheGet, cacheInvalidatePrefixes, cacheSet, throttleTake)
|
||
|
|
import Nuthatch.S3
|
||
|
|
( ListPage (..)
|
||
|
|
, ObjectInfo (..)
|
||
|
|
, S3
|
||
|
|
, deleteObject
|
||
|
|
, getObjectBytes
|
||
|
|
, headObject
|
||
|
|
, listObjects
|
||
|
|
, putObject
|
||
|
|
)
|
||
|
|
import Nuthatch.Session
|
||
|
|
import Nuthatch.Types
|
||
|
|
import Nuthatch.Ui
|
||
|
|
|
||
|
|
data Env = Env
|
||
|
|
{ envConfig :: Config
|
||
|
|
, envDb :: Db
|
||
|
|
, envRedis :: Redis
|
||
|
|
, envS3 :: S3
|
||
|
|
}
|
||
|
|
|
||
|
|
newEnv :: Config -> Db -> Redis -> S3 -> Env
|
||
|
|
newEnv = Env
|
||
|
|
|
||
|
|
application :: Env -> Application
|
||
|
|
application env req respond =
|
||
|
|
case (requestMethod req, pathInfo req) of
|
||
|
|
("GET", ["healthz"]) ->
|
||
|
|
respond (json status200 (A.object ["ok" A..= True]))
|
||
|
|
("GET", ["readyz"]) ->
|
||
|
|
respond =<< readiness env
|
||
|
|
("GET", ["login"]) ->
|
||
|
|
respond (htmlResponse status200 (loginPage ""))
|
||
|
|
("POST", ["login"]) ->
|
||
|
|
handleLogin env req respond
|
||
|
|
("GET", ["logout"]) ->
|
||
|
|
handleLogout env respond
|
||
|
|
("POST", ["logout"]) ->
|
||
|
|
handleLogout env respond
|
||
|
|
("GET", []) ->
|
||
|
|
withSession env req respond handleBrowse
|
||
|
|
("GET", ["download"]) ->
|
||
|
|
withSession env req respond handleDownload
|
||
|
|
("POST", ["upload"]) ->
|
||
|
|
withSession env req respond handleUpload
|
||
|
|
("POST", ["delete"]) ->
|
||
|
|
withSession env req respond handleDelete
|
||
|
|
_ ->
|
||
|
|
respond (text status404 "not found")
|
||
|
|
|
||
|
|
readiness :: Env -> IO Response
|
||
|
|
readiness env = do
|
||
|
|
pg <- try $ Db.query_ (envDb env) "SELECT 1" :: IO (Either SomeException [Only Int])
|
||
|
|
pure $ case pg of
|
||
|
|
Right _ -> json status200 (A.object ["ready" A..= True])
|
||
|
|
Left _ -> json status503 (A.object ["ready" A..= False])
|
||
|
|
|
||
|
|
handleLogin :: Env -> Request -> (Response -> IO ResponseReceived) -> IO ResponseReceived
|
||
|
|
handleLogin env req respond = do
|
||
|
|
(params, _) <- parseRequestBody lbsBackEnd req
|
||
|
|
let username =
|
||
|
|
T.strip . TE.decodeUtf8 . fromMaybe "" $
|
||
|
|
lookup "username" params
|
||
|
|
if T.null username || T.any (`elem` (" \t\r\n" :: String)) username
|
||
|
|
then respond (htmlResponse status400 (loginPage "Enter a single username token."))
|
||
|
|
else do
|
||
|
|
let uid = encodeUserId "nuthatch-dev" username
|
||
|
|
ensurePersonalPool (envDb env) (envConfig env) uid
|
||
|
|
tokenResult <- mintSession (envConfig env) "nuthatch-dev" username (Just username)
|
||
|
|
csrf <- newCsrfToken
|
||
|
|
case tokenResult of
|
||
|
|
Left err -> respond (htmlResponse status500 (loginPage err))
|
||
|
|
Right jwt -> do
|
||
|
|
audit "login" "ok" (Just (unUserId uid)) []
|
||
|
|
respond $
|
||
|
|
responseLBS status302
|
||
|
|
[ (hLocation, "/")
|
||
|
|
, setCookie (envConfig env) (cfgCookieName (envConfig env)) jwt True
|
||
|
|
, setCookie (envConfig env) (cfgCsrfCookieName (envConfig env)) csrf False
|
||
|
|
]
|
||
|
|
""
|
||
|
|
|
||
|
|
handleLogout :: Env -> (Response -> IO ResponseReceived) -> IO ResponseReceived
|
||
|
|
handleLogout env respond =
|
||
|
|
respond $
|
||
|
|
responseLBS status302
|
||
|
|
[ (hLocation, "/login")
|
||
|
|
, clearCookie (envConfig env) (cfgCookieName (envConfig env))
|
||
|
|
, clearCookie (envConfig env) (cfgCsrfCookieName (envConfig env))
|
||
|
|
]
|
||
|
|
""
|
||
|
|
|
||
|
|
withSession
|
||
|
|
:: Env
|
||
|
|
-> Request
|
||
|
|
-> (Response -> IO ResponseReceived)
|
||
|
|
-> (Env -> Request -> SessionUser -> Text -> (Response -> IO ResponseReceived) -> IO ResponseReceived)
|
||
|
|
-> IO ResponseReceived
|
||
|
|
withSession env req respond handler = do
|
||
|
|
let cookies = parseCookies (requestHeaders req)
|
||
|
|
cfg = envConfig env
|
||
|
|
case lookup (TE.encodeUtf8 (cfgCookieName cfg)) cookies of
|
||
|
|
Nothing -> respond (redirect "/login")
|
||
|
|
Just raw -> do
|
||
|
|
verified <- verifySession cfg (TE.decodeUtf8 raw)
|
||
|
|
case verified of
|
||
|
|
Left _ -> respond (redirect "/login")
|
||
|
|
Right user -> do
|
||
|
|
csrf <- case lookup (TE.encodeUtf8 (cfgCsrfCookieName cfg)) cookies of
|
||
|
|
Just c -> pure (TE.decodeUtf8 c)
|
||
|
|
Nothing -> newCsrfToken
|
||
|
|
handler env req user csrf respond
|
||
|
|
|
||
|
|
handleBrowse
|
||
|
|
:: Env
|
||
|
|
-> Request
|
||
|
|
-> SessionUser
|
||
|
|
-> Text
|
||
|
|
-> (Response -> IO ResponseReceived)
|
||
|
|
-> IO ResponseReceived
|
||
|
|
handleBrowse env req user csrf respond = do
|
||
|
|
let qs = queryString req
|
||
|
|
root = fromMaybe Personal (parseRoot =<< (TE.decodeUtf8 <$> join (lookup "root" qs)))
|
||
|
|
prefix = maybe "" TE.decodeUtf8 (join (lookup "prefix" qs))
|
||
|
|
q = maybe "" TE.decodeUtf8 (join (lookup "q" qs))
|
||
|
|
notice = maybe "" TE.decodeUtf8 (join (lookup "notice" qs))
|
||
|
|
errQ = maybe "" TE.decodeUtf8 (join (lookup "error" qs))
|
||
|
|
ensurePersonalPool (envDb env) (envConfig env) (suUserId user)
|
||
|
|
entriesResult <-
|
||
|
|
if T.null q
|
||
|
|
then listEntries env user root prefix
|
||
|
|
else searchEntries env user root q
|
||
|
|
uploadOp <- toText <$> nextRandom
|
||
|
|
case entriesResult of
|
||
|
|
Left err ->
|
||
|
|
respond
|
||
|
|
(htmlResponse status500
|
||
|
|
(browsePage user root prefix csrf uploadOp [] [] (Just err) notice))
|
||
|
|
Right entries -> do
|
||
|
|
deleteOps <- mapM (\_ -> toText <$> nextRandom) entries
|
||
|
|
let mErr = if T.null errQ then Nothing else Just errQ
|
||
|
|
respond
|
||
|
|
(htmlResponse status200
|
||
|
|
(browsePage user root prefix csrf uploadOp deleteOps entries mErr notice))
|
||
|
|
|
||
|
|
handleDownload
|
||
|
|
:: Env
|
||
|
|
-> Request
|
||
|
|
-> SessionUser
|
||
|
|
-> Text
|
||
|
|
-> (Response -> IO ResponseReceived)
|
||
|
|
-> IO ResponseReceived
|
||
|
|
handleDownload env req user _csrf respond = do
|
||
|
|
let qs = queryString req
|
||
|
|
mRoot = parseRoot =<< (TE.decodeUtf8 <$> join (lookup "root" qs))
|
||
|
|
mPath = TE.decodeUtf8 <$> join (lookup "path" qs)
|
||
|
|
case (mRoot, mPath) of
|
||
|
|
(Just root, Just rawPath) ->
|
||
|
|
case validateRelativePath rawPath of
|
||
|
|
Left err -> respond (text status400 err)
|
||
|
|
Right rel -> do
|
||
|
|
let key = buildObjectKey root (suUserId user) rel
|
||
|
|
meta <- headObject (envS3 env) key
|
||
|
|
case meta of
|
||
|
|
Left err -> respond (text status502 err)
|
||
|
|
Right Nothing -> respond (text status404 "not found")
|
||
|
|
Right (Just info) -> do
|
||
|
|
allowed <- throttleTake (envRedis env) (envConfig env) (suUserId user) (oiSize info)
|
||
|
|
case allowed of
|
||
|
|
Left _ -> respond (text status503 "bandwidth throttle unavailable")
|
||
|
|
Right False -> respond (text status429 "bandwidth limit exceeded")
|
||
|
|
Right True -> do
|
||
|
|
bodyResult <- getObjectBytes (envS3 env) key
|
||
|
|
case bodyResult of
|
||
|
|
Left err -> respond (text status502 err)
|
||
|
|
Right body -> do
|
||
|
|
audit "download" "ok" (Just (unUserId (suUserId user)))
|
||
|
|
[ ("root", A.toJSON (rootLabel root))
|
||
|
|
, ("bytes", A.toJSON (oiSize info))
|
||
|
|
]
|
||
|
|
let filename = basename rel
|
||
|
|
disp =
|
||
|
|
"attachment; filename=\""
|
||
|
|
<> TE.encodeUtf8 (sanitizeFilename filename)
|
||
|
|
<> "\""
|
||
|
|
respond $
|
||
|
|
responseLBS status200
|
||
|
|
[ (hContentType, "application/octet-stream")
|
||
|
|
, (hContentDisposition, disp)
|
||
|
|
, ("X-Content-Type-Options", "nosniff")
|
||
|
|
]
|
||
|
|
body
|
||
|
|
_ -> respond (text status400 "root and path are required")
|
||
|
|
|
||
|
|
handleUpload
|
||
|
|
:: Env
|
||
|
|
-> Request
|
||
|
|
-> SessionUser
|
||
|
|
-> Text
|
||
|
|
-> (Response -> IO ResponseReceived)
|
||
|
|
-> IO ResponseReceived
|
||
|
|
handleUpload env req user csrf respond = do
|
||
|
|
(params, files) <- parseRequestBody lbsBackEnd req
|
||
|
|
let providedCsrf = maybe "" TE.decodeUtf8 (lookup "csrf" params)
|
||
|
|
if not (constantTimeEq csrf providedCsrf)
|
||
|
|
then respond (text status403 "csrf validation failed")
|
||
|
|
else do
|
||
|
|
let root = fromMaybe Personal (parseRoot . TE.decodeUtf8 =<< lookup "root" params)
|
||
|
|
prefix = maybe "" TE.decodeUtf8 (lookup "prefix" params)
|
||
|
|
case lookup "file" files of
|
||
|
|
Nothing ->
|
||
|
|
respond (redirect (browseUrl root prefix "" "missing file"))
|
||
|
|
Just fileInfo -> do
|
||
|
|
let name = TE.decodeUtf8 (fileName fileInfo)
|
||
|
|
content = fileContent fileInfo
|
||
|
|
size = toInteger (BL.length content)
|
||
|
|
if size <= 0
|
||
|
|
then respond (redirect (browseUrl root prefix "" "empty file"))
|
||
|
|
else if size > cfgMaxObjectBytes (envConfig env)
|
||
|
|
then respond (text status413 "object too large")
|
||
|
|
else case validateRelativePath (joinPath prefix name) of
|
||
|
|
Left err -> respond (redirect (browseUrl root prefix "" err))
|
||
|
|
Right rel -> do
|
||
|
|
allowed <- throttleTake (envRedis env) (envConfig env) (suUserId user) size
|
||
|
|
case allowed of
|
||
|
|
Left _ -> respond (text status503 "bandwidth throttle unavailable")
|
||
|
|
Right False -> respond (text status429 "bandwidth limit exceeded")
|
||
|
|
Right True -> do
|
||
|
|
let sha = contentSha256 content
|
||
|
|
op <- parseOperationId params
|
||
|
|
outcome <- runPut env user root rel op size sha content
|
||
|
|
case outcome of
|
||
|
|
Left err ->
|
||
|
|
respond (redirect (browseUrl root prefix "" err))
|
||
|
|
Right MutationSucceeded -> do
|
||
|
|
audit "put" "ok" (Just (unUserId (suUserId user)))
|
||
|
|
[ ("root", A.toJSON (rootLabel root))
|
||
|
|
, ("bytes", A.toJSON size)
|
||
|
|
, ("operation_id", A.toJSON (unOperationId op))
|
||
|
|
]
|
||
|
|
respond (redirect (browseUrl root prefix "uploaded" ""))
|
||
|
|
Right MutationRejected ->
|
||
|
|
respond (redirect (browseUrl root prefix "" "upload rejected (quota or lock)"))
|
||
|
|
Right MutationConflict ->
|
||
|
|
respond (redirect (browseUrl root prefix "" "operation conflict"))
|
||
|
|
Right MutationPending ->
|
||
|
|
respond (redirect (browseUrl root prefix "" "upload still pending; retry same operation id"))
|
||
|
|
Right other ->
|
||
|
|
respond (redirect (browseUrl root prefix "" ("upload " <> T.pack (show other))))
|
||
|
|
|
||
|
|
handleDelete
|
||
|
|
:: Env
|
||
|
|
-> Request
|
||
|
|
-> SessionUser
|
||
|
|
-> Text
|
||
|
|
-> (Response -> IO ResponseReceived)
|
||
|
|
-> IO ResponseReceived
|
||
|
|
handleDelete env req user csrf respond = do
|
||
|
|
(params, _) <- parseRequestBody lbsBackEnd req
|
||
|
|
let providedCsrf = maybe "" TE.decodeUtf8 (lookup "csrf" params)
|
||
|
|
if not (constantTimeEq csrf providedCsrf)
|
||
|
|
then respond (text status403 "csrf validation failed")
|
||
|
|
else do
|
||
|
|
let root = fromMaybe Personal (parseRoot . TE.decodeUtf8 =<< lookup "root" params)
|
||
|
|
rawPath = maybe "" TE.decodeUtf8 (lookup "path" params)
|
||
|
|
isPrefix = maybe False (== "1") (TE.decodeUtf8 <$> lookup "prefix_delete" params)
|
||
|
|
parent = parentOf rawPath
|
||
|
|
if isPrefix
|
||
|
|
then do
|
||
|
|
result <- deletePrefix env user root rawPath
|
||
|
|
case result of
|
||
|
|
Left err -> respond (redirect (browseUrl root parent "" err))
|
||
|
|
Right n ->
|
||
|
|
respond
|
||
|
|
(redirect
|
||
|
|
(browseUrl root parent ("deleted " <> T.pack (show n) <> " object(s)") ""))
|
||
|
|
else
|
||
|
|
case validateRelativePath rawPath of
|
||
|
|
Left err -> respond (redirect (browseUrl root parent "" err))
|
||
|
|
Right rel -> do
|
||
|
|
op <- parseOperationId params
|
||
|
|
outcome <- runDelete env user root rel op
|
||
|
|
case outcome of
|
||
|
|
Left err -> respond (redirect (browseUrl root parent "" err))
|
||
|
|
Right MutationSucceeded -> do
|
||
|
|
audit "delete" "ok" (Just (unUserId (suUserId user)))
|
||
|
|
[ ("root", A.toJSON (rootLabel root))
|
||
|
|
, ("operation_id", A.toJSON (unOperationId op))
|
||
|
|
]
|
||
|
|
respond (redirect (browseUrl root parent "deleted" ""))
|
||
|
|
Right MutationRejected ->
|
||
|
|
respond (redirect (browseUrl root parent "" "delete rejected"))
|
||
|
|
Right MutationPending ->
|
||
|
|
respond (redirect (browseUrl root parent "" "delete still pending; retry same operation id"))
|
||
|
|
Right other ->
|
||
|
|
respond (redirect (browseUrl root parent "" ("delete " <> T.pack (show other))))
|
||
|
|
|
||
|
|
runPut
|
||
|
|
:: Env
|
||
|
|
-> SessionUser
|
||
|
|
-> Root
|
||
|
|
-> RelativePath
|
||
|
|
-> OperationId
|
||
|
|
-> Integer
|
||
|
|
-> ContentSha256
|
||
|
|
-> BL.ByteString
|
||
|
|
-> IO (Either Text MutationResult)
|
||
|
|
runPut env user root rel op size sha body = do
|
||
|
|
ensurePersonalPool (envDb env) (envConfig env) (suUserId user)
|
||
|
|
let uid = suUserId user
|
||
|
|
key = buildObjectKey root uid rel
|
||
|
|
pool = case root of
|
||
|
|
Personal -> personalPool uid
|
||
|
|
Shared -> sharedPool
|
||
|
|
outcome <- withTransaction (envDb env) $ \conn ->
|
||
|
|
admitMutation conn uid Put op pool key rel size (Just sha)
|
||
|
|
(cfgMaxObjectBytes (envConfig env))
|
||
|
|
case outcome of
|
||
|
|
AdmitConflict -> pure (Right MutationConflict)
|
||
|
|
AdmitRejected reason _ -> pure (Left reason)
|
||
|
|
AdmitReplay MutationPending _ -> pure (Right MutationPending)
|
||
|
|
AdmitReplay result _ -> pure (Right result)
|
||
|
|
AdmitPrepared rec -> dispatchAndFinalize env user root rec $ do
|
||
|
|
_ <- putObject (envS3 env) key size sha body
|
||
|
|
pure ()
|
||
|
|
|
||
|
|
runDelete
|
||
|
|
:: Env
|
||
|
|
-> SessionUser
|
||
|
|
-> Root
|
||
|
|
-> RelativePath
|
||
|
|
-> OperationId
|
||
|
|
-> IO (Either Text MutationResult)
|
||
|
|
runDelete env user root rel op = do
|
||
|
|
ensurePersonalPool (envDb env) (envConfig env) (suUserId user)
|
||
|
|
let uid = suUserId user
|
||
|
|
key = buildObjectKey root uid rel
|
||
|
|
pool = case root of
|
||
|
|
Personal -> personalPool uid
|
||
|
|
Shared -> sharedPool
|
||
|
|
outcome <- withTransaction (envDb env) $ \conn ->
|
||
|
|
admitMutation conn uid Delete op pool key rel 0 Nothing
|
||
|
|
(cfgMaxObjectBytes (envConfig env))
|
||
|
|
case outcome of
|
||
|
|
AdmitConflict -> pure (Right MutationConflict)
|
||
|
|
AdmitRejected reason _ -> pure (Left reason)
|
||
|
|
AdmitReplay MutationPending _ -> pure (Right MutationPending)
|
||
|
|
AdmitReplay result _ -> pure (Right result)
|
||
|
|
AdmitPrepared rec -> dispatchAndFinalize env user root rec $ do
|
||
|
|
_ <- deleteObject (envS3 env) key
|
||
|
|
pure ()
|
||
|
|
|
||
|
|
dispatchAndFinalize
|
||
|
|
:: Env
|
||
|
|
-> SessionUser
|
||
|
|
-> Root
|
||
|
|
-> MutationRecord
|
||
|
|
-> IO ()
|
||
|
|
-> IO (Either Text MutationResult)
|
||
|
|
dispatchAndFinalize env user root rec dispatch = do
|
||
|
|
prepared <- withTransaction (envDb env) $ \conn -> do
|
||
|
|
moved <- markApplying conn rec
|
||
|
|
if not moved
|
||
|
|
then pure False
|
||
|
|
else markExternalPending conn rec
|
||
|
|
if not prepared
|
||
|
|
then pure (Right MutationPending)
|
||
|
|
else do
|
||
|
|
_ <- try dispatch :: IO (Either SomeException ())
|
||
|
|
observed <- observeObject (envS3 env) (mrObjectKey rec)
|
||
|
|
finalized <- withTransaction (envDb env) $ \conn -> do
|
||
|
|
settled <- markExternalSettled conn rec
|
||
|
|
if not settled
|
||
|
|
then pure Nothing
|
||
|
|
else finalizeFromObservation conn rec observed
|
||
|
|
case finalized of
|
||
|
|
Just MutationSucceeded -> do
|
||
|
|
cacheInvalidatePrefixes (envRedis env) (suUserId user) root
|
||
|
|
(listCacheAncestorPrefixes (mrRelativePath rec))
|
||
|
|
pure (Right MutationSucceeded)
|
||
|
|
Just MutationFailed ->
|
||
|
|
pure (Left "mutation failed after store settlement")
|
||
|
|
Just other -> pure (Right other)
|
||
|
|
Nothing -> pure (Right MutationPending)
|
||
|
|
|
||
|
|
observeObject :: S3 -> ObjectKey -> IO ObjectObservation
|
||
|
|
observeObject s3 key = do
|
||
|
|
result <- headObject s3 key
|
||
|
|
pure $ case result of
|
||
|
|
Right (Just info) -> ObjectPresent (oiSize info) (oiSha256 info)
|
||
|
|
Right Nothing -> ObjectAbsent
|
||
|
|
Left err -> ObjectUnavailable err
|
||
|
|
|
||
|
|
finalizeFromObservation
|
||
|
|
:: Connection
|
||
|
|
-> MutationRecord
|
||
|
|
-> ObjectObservation
|
||
|
|
-> IO (Maybe MutationResult)
|
||
|
|
finalizeFromObservation conn rec observed =
|
||
|
|
case observed of
|
||
|
|
ObjectUnavailable _ -> pure Nothing
|
||
|
|
_ | desiredMatches rec observed -> do
|
||
|
|
let (finalSize, finalSha) =
|
||
|
|
case mrKind rec of
|
||
|
|
Put ->
|
||
|
|
( mrDeclaredSize rec
|
||
|
|
, mrContentSha256 rec
|
||
|
|
)
|
||
|
|
Delete -> (0, Nothing)
|
||
|
|
ok <- finalizeSuccess conn rec finalSize finalSha
|
||
|
|
pure $ if ok then Just MutationSucceeded else Nothing
|
||
|
|
_ | priorMatches rec observed -> do
|
||
|
|
ok <- finalizeFailure conn rec
|
||
|
|
pure $ if ok then Just MutationFailed else Nothing
|
||
|
|
_ -> do
|
||
|
|
ok <- finalizeFailure conn rec
|
||
|
|
pure $ if ok then Just MutationFailed else Nothing
|
||
|
|
|
||
|
|
deletePrefix :: Env -> SessionUser -> Root -> Text -> IO (Either Text Int)
|
||
|
|
deletePrefix env user root prefix = do
|
||
|
|
let listPrefix =
|
||
|
|
case root of
|
||
|
|
Personal ->
|
||
|
|
"users/" <> unUserId (suUserId user) <> "/" <> withSlash prefix
|
||
|
|
Shared ->
|
||
|
|
"shared/" <> withSlash prefix
|
||
|
|
keysResult <- collectKeys (envS3 env) listPrefix
|
||
|
|
case keysResult of
|
||
|
|
Left err -> pure (Left err)
|
||
|
|
Right keys -> do
|
||
|
|
let rootPrefix =
|
||
|
|
case root of
|
||
|
|
Personal -> "users/" <> unUserId (suUserId user) <> "/"
|
||
|
|
Shared -> "shared/"
|
||
|
|
rels =
|
||
|
|
[ RelativePath rel
|
||
|
|
| k <- keys
|
||
|
|
, T.isPrefixOf rootPrefix k
|
||
|
|
, let rel = T.drop (T.length rootPrefix) k
|
||
|
|
, not (T.null rel)
|
||
|
|
]
|
||
|
|
results <- mapM
|
||
|
|
(\rel -> do
|
||
|
|
op <- OperationId . toText <$> nextRandom
|
||
|
|
runDelete env user root rel op)
|
||
|
|
rels
|
||
|
|
let ok = length [ () | Right MutationSucceeded <- results ]
|
||
|
|
err = listToMaybe [ e | Left e <- results ]
|
||
|
|
pure $ maybe (Right ok) Left err
|
||
|
|
where
|
||
|
|
withSlash p
|
||
|
|
| T.null p = ""
|
||
|
|
| T.isSuffixOf "/" p = p
|
||
|
|
| otherwise = p <> "/"
|
||
|
|
|
||
|
|
collectKeys :: S3 -> Text -> IO (Either Text [Text])
|
||
|
|
collectKeys s3 prefix = go Nothing []
|
||
|
|
where
|
||
|
|
go mTok acc = do
|
||
|
|
page <- listObjects s3 prefix Nothing mTok 1000
|
||
|
|
case page of
|
||
|
|
Left err -> pure (Left err)
|
||
|
|
Right lp -> do
|
||
|
|
let keys = map fst (lpObjects lp)
|
||
|
|
acc' = acc <> keys
|
||
|
|
if lpIsTruncated lp
|
||
|
|
then go (lpContinuationToken lp) acc'
|
||
|
|
else pure (Right acc')
|
||
|
|
|
||
|
|
listEntries :: Env -> SessionUser -> Root -> Text -> IO (Either Text [ListEntry])
|
||
|
|
listEntries env user root prefix = do
|
||
|
|
let cacheKey =
|
||
|
|
"nuthatch:list:"
|
||
|
|
<> unUserId (suUserId user)
|
||
|
|
<> ":"
|
||
|
|
<> rootLabel root
|
||
|
|
<> ":"
|
||
|
|
<> prefix
|
||
|
|
<> ":"
|
||
|
|
s3Prefix =
|
||
|
|
case root of
|
||
|
|
Personal -> "users/" <> unUserId (suUserId user) <> "/" <> prefixSlash
|
||
|
|
Shared -> "shared/" <> prefixSlash
|
||
|
|
prefixSlash
|
||
|
|
| T.null prefix = ""
|
||
|
|
| T.isSuffixOf "/" prefix = prefix
|
||
|
|
| otherwise = prefix <> "/"
|
||
|
|
cached <- cacheGet (envRedis env) cacheKey
|
||
|
|
case cached >>= A.decodeStrict of
|
||
|
|
Just entries -> pure (Right entries)
|
||
|
|
Nothing -> do
|
||
|
|
page <- listObjects (envS3 env) s3Prefix (Just "/") Nothing 1000
|
||
|
|
case page of
|
||
|
|
Left err -> pure (Left err)
|
||
|
|
Right lp -> do
|
||
|
|
let rootPrefix =
|
||
|
|
case root of
|
||
|
|
Personal -> "users/" <> unUserId (suUserId user) <> "/"
|
||
|
|
Shared -> "shared/"
|
||
|
|
folders =
|
||
|
|
[ ListEntry
|
||
|
|
{ leName = lastSeg (T.dropEnd 1 (T.drop (T.length rootPrefix) p))
|
||
|
|
, leIsPrefix = True
|
||
|
|
, leSize = Nothing
|
||
|
|
, leRelativePath = T.dropEnd 1 (T.drop (T.length rootPrefix) p)
|
||
|
|
}
|
||
|
|
| p <- lpPrefixes lp
|
||
|
|
, T.isPrefixOf rootPrefix p
|
||
|
|
]
|
||
|
|
files =
|
||
|
|
[ ListEntry
|
||
|
|
{ leName = lastSeg rel
|
||
|
|
, leIsPrefix = False
|
||
|
|
, leSize = Just sz
|
||
|
|
, leRelativePath = rel
|
||
|
|
}
|
||
|
|
| (k, sz) <- lpObjects lp
|
||
|
|
, T.isPrefixOf rootPrefix k
|
||
|
|
, let rel = T.drop (T.length rootPrefix) k
|
||
|
|
, not (T.null rel)
|
||
|
|
, not (T.isSuffixOf "/" rel)
|
||
|
|
]
|
||
|
|
entries = folders <> files
|
||
|
|
cacheSet (envRedis env) cacheKey (cfgListCacheTtlSec (envConfig env))
|
||
|
|
(BL.toStrict (A.encode entries))
|
||
|
|
pure (Right entries)
|
||
|
|
|
||
|
|
searchEntries :: Env -> SessionUser -> Root -> Text -> IO (Either Text [ListEntry])
|
||
|
|
searchEntries env user root q = do
|
||
|
|
let s3Prefix =
|
||
|
|
case root of
|
||
|
|
Personal -> "users/" <> unUserId (suUserId user) <> "/"
|
||
|
|
Shared -> "shared/"
|
||
|
|
needle = T.toLower q
|
||
|
|
keysResult <- collectKeysBounded (envS3 env) s3Prefix (cfgSearchMaxKeys (envConfig env))
|
||
|
|
pure $ case keysResult of
|
||
|
|
Left err -> Left err
|
||
|
|
Right keys ->
|
||
|
|
Right
|
||
|
|
[ ListEntry
|
||
|
|
{ leName = lastSeg rel
|
||
|
|
, leIsPrefix = False
|
||
|
|
, leSize = Just sz
|
||
|
|
, leRelativePath = rel
|
||
|
|
}
|
||
|
|
| (k, sz) <- keys
|
||
|
|
, T.isPrefixOf s3Prefix k
|
||
|
|
, let rel = T.drop (T.length s3Prefix) k
|
||
|
|
, not (T.null rel)
|
||
|
|
, needle `T.isInfixOf` T.toLower rel
|
||
|
|
]
|
||
|
|
|
||
|
|
collectKeysBounded :: S3 -> Text -> Int -> IO (Either Text [(Text, Integer)])
|
||
|
|
collectKeysBounded s3 prefix maxKeys = go Nothing [] 0
|
||
|
|
where
|
||
|
|
go mTok acc n
|
||
|
|
| n >= maxKeys = pure (Right (take maxKeys acc))
|
||
|
|
| otherwise = do
|
||
|
|
page <- listObjects s3 prefix Nothing mTok (min 1000 (maxKeys - n))
|
||
|
|
case page of
|
||
|
|
Left err -> pure (Left err)
|
||
|
|
Right lp -> do
|
||
|
|
let acc' = acc <> lpObjects lp
|
||
|
|
n' = length acc'
|
||
|
|
if lpIsTruncated lp && n' < maxKeys
|
||
|
|
then go (lpContinuationToken lp) acc' n'
|
||
|
|
else pure (Right (take maxKeys acc'))
|
||
|
|
|
||
|
|
runRecovery :: Env -> IO ()
|
||
|
|
runRecovery env = forever $ do
|
||
|
|
_ <- try (recoverOnce env) :: IO (Either SomeException ())
|
||
|
|
threadDelay (5 * 1000000)
|
||
|
|
|
||
|
|
recoverOnce :: Env -> IO ()
|
||
|
|
recoverOnce env = do
|
||
|
|
let grace = cfgRecoveryGraceSec (envConfig env)
|
||
|
|
batch <- withTransaction (envDb env) $ \conn -> claimActiveBatch conn 20 grace
|
||
|
|
mapM_ (recoverOne env) batch
|
||
|
|
|
||
|
|
recoverOne :: Env -> MutationRecord -> IO ()
|
||
|
|
recoverOne env rec =
|
||
|
|
case (mrState rec, mrExternal rec) of
|
||
|
|
(Prepared, _) -> do
|
||
|
|
_ <- withTransaction (envDb env) $ \conn -> do
|
||
|
|
moved <- markApplying conn rec
|
||
|
|
when moved $ void (markSettledFromIdle conn rec)
|
||
|
|
observed <- observeObject (envS3 env) (mrObjectKey rec)
|
||
|
|
_ <- withTransaction (envDb env) $ \conn ->
|
||
|
|
finalizeFromObservation conn rec observed
|
||
|
|
pure ()
|
||
|
|
(Applying, ExtIdle) -> do
|
||
|
|
_ <- withTransaction (envDb env) $ \conn -> markSettledFromIdle conn rec
|
||
|
|
observed <- observeObject (envS3 env) (mrObjectKey rec)
|
||
|
|
_ <- withTransaction (envDb env) $ \conn ->
|
||
|
|
finalizeFromObservation conn rec observed
|
||
|
|
pure ()
|
||
|
|
(Applying, ExtPending) -> do
|
||
|
|
observed <- observeObject (envS3 env) (mrObjectKey rec)
|
||
|
|
_ <- withTransaction (envDb env) $ \conn -> do
|
||
|
|
settled <- markExternalSettled conn rec
|
||
|
|
if settled
|
||
|
|
then finalizeFromObservation conn rec observed
|
||
|
|
else pure Nothing
|
||
|
|
pure ()
|
||
|
|
(Applying, ExtSettled) -> do
|
||
|
|
observed <- observeObject (envS3 env) (mrObjectKey rec)
|
||
|
|
_ <- withTransaction (envDb env) $ \conn ->
|
||
|
|
finalizeFromObservation conn rec observed
|
||
|
|
pure ()
|
||
|
|
_ -> pure ()
|
||
|
|
|
||
|
|
parseOperationId :: [(BS.ByteString, BS.ByteString)] -> IO OperationId
|
||
|
|
parseOperationId params =
|
||
|
|
case lookup "operation_id" params of
|
||
|
|
Just raw | not (BS.null raw) ->
|
||
|
|
pure (OperationId (TE.decodeUtf8 raw))
|
||
|
|
_ ->
|
||
|
|
OperationId . toText <$> nextRandom
|
||
|
|
|
||
|
|
joinPath :: Text -> Text -> Text
|
||
|
|
joinPath prefix name
|
||
|
|
| T.null prefix = name
|
||
|
|
| T.isSuffixOf "/" prefix = prefix <> name
|
||
|
|
| otherwise = prefix <> "/" <> name
|
||
|
|
|
||
|
|
parentOf :: Text -> Text
|
||
|
|
parentOf path =
|
||
|
|
case reverse (T.splitOn "/" path) of
|
||
|
|
(_ : rest) -> T.intercalate "/" (reverse rest)
|
||
|
|
_ -> ""
|
||
|
|
|
||
|
|
browseUrl :: Root -> Text -> Text -> Text -> Text
|
||
|
|
browseUrl root prefix notice err =
|
||
|
|
"/?root=" <> rootLabel root
|
||
|
|
<> (if T.null prefix then "" else "&prefix=" <> prefix)
|
||
|
|
<> (if T.null notice then "" else "¬ice=" <> notice)
|
||
|
|
<> (if T.null err then "" else "&error=" <> err)
|
||
|
|
|
||
|
|
contentSha256 :: BL.ByteString -> ContentSha256
|
||
|
|
contentSha256 body =
|
||
|
|
ContentSha256 . TE.decodeUtf8 . B16.encode . BA.convert $
|
||
|
|
(hashlazy body :: Digest SHA256)
|
||
|
|
|
||
|
|
sanitizeFilename :: Text -> Text
|
||
|
|
sanitizeFilename =
|
||
|
|
T.map $ \c -> if c `elem` ("\"\\\r\n" :: String) then '_' else c
|
||
|
|
|
||
|
|
lastSeg :: Text -> Text
|
||
|
|
lastSeg t =
|
||
|
|
case reverse (T.splitOn "/" t) of
|
||
|
|
(x : _) -> x
|
||
|
|
[] -> t
|
||
|
|
|
||
|
|
json :: Status -> A.Value -> Response
|
||
|
|
json st v =
|
||
|
|
responseLBS st [(hContentType, "application/json")] (A.encode v)
|
||
|
|
|
||
|
|
text :: Status -> Text -> Response
|
||
|
|
text st t =
|
||
|
|
responseLBS st [(hContentType, "text/plain; charset=utf-8")]
|
||
|
|
(BL.fromStrict (TE.encodeUtf8 t))
|
||
|
|
|
||
|
|
redirect :: Text -> Response
|
||
|
|
redirect loc =
|
||
|
|
responseLBS status302 [(hLocation, TE.encodeUtf8 loc)] ""
|
||
|
|
|
||
|
|
setCookie :: Config -> Text -> Text -> Bool -> (HeaderName, BS.ByteString)
|
||
|
|
setCookie _cfg name value httpOnly =
|
||
|
|
( "Set-Cookie"
|
||
|
|
, TE.encodeUtf8 $
|
||
|
|
name <> "=" <> value
|
||
|
|
<> "; Path=/"
|
||
|
|
<> "; SameSite=Lax"
|
||
|
|
<> (if httpOnly then "; HttpOnly" else "")
|
||
|
|
)
|
||
|
|
|
||
|
|
clearCookie :: Config -> Text -> (HeaderName, BS.ByteString)
|
||
|
|
clearCookie _cfg name =
|
||
|
|
( "Set-Cookie"
|
||
|
|
, TE.encodeUtf8 (name <> "=; Path=/; Max-Age=0; SameSite=Lax")
|
||
|
|
)
|
||
|
|
|
||
|
|
parseCookies :: [Header] -> [(BS.ByteString, BS.ByteString)]
|
||
|
|
parseCookies headers =
|
||
|
|
case lookup "Cookie" headers of
|
||
|
|
Nothing -> []
|
||
|
|
Just raw ->
|
||
|
|
[ (trimBS k, trimBS (BS8.drop 1 v))
|
||
|
|
| part <- BS8.split ';' raw
|
||
|
|
, let (k, v) = BS8.break (== '=') part
|
||
|
|
, not (BS.null v)
|
||
|
|
]
|
||
|
|
|
||
|
|
trimBS :: BS.ByteString -> BS.ByteString
|
||
|
|
trimBS = BS8.dropWhile isSpace . BS8.dropWhileEnd isSpace
|
||
|
|
where
|
||
|
|
isSpace c = c == ' ' || c == '\t'
|
||
|
|
|
||
|
|
join :: Maybe (Maybe a) -> Maybe a
|
||
|
|
join = fromMaybe Nothing
|
||
|
|
|
||
|
|
status502 :: Status
|
||
|
|
status502 = Status 502 "Bad Gateway"
|