ms2k/nuthatch
SHA256
Watch
0
Code Releases Activity

Group configuration by component ownership

Browse source
This commit is contained in:
ms2k 2026-07-20 02:21:33 +09:00
commit 67cc0cb802
27 changed files with 486 additions and 258 deletions
Download patch file Download diff file Expand all files Collapse all files

34
app/Main.hs
View file

@ -24,34 +24,40 @@ main :: IO ()
main = do
hSetBuffering stdout LineBuffering
cfg <- loadConfig
let runtimeCfg = cfgRuntime cfg
httpCfg = cfgHttp cfg
postgresCfg = cfgPostgres cfg
redisCfg = cfgRedis cfg
s3Cfg = cfgS3 cfg
productCfg = cfgProduct cfg
retryStartup
(cfgStartupAttempts cfg)
(cfgStartupDelayMs cfg)
(cfgStartupAttempts runtimeCfg)
(cfgStartupDelayMs runtimeCfg)
"postgresql"
(withDb cfg $ \db -> do
(withDb postgresCfg $ \db -> do
_ <- query_ db "SELECT 1" :: IO [Only Int]
pure ())
withDb cfg $ \db -> do
withDb postgresCfg $ \db -> do
schemaPath <- findSchema
retryStartup
(cfgStartupAttempts cfg)
(cfgStartupDelayMs cfg)
(cfgStartupAttempts runtimeCfg)
(cfgStartupDelayMs runtimeCfg)
"migration"
(migrate db schemaPath cfg)
withRedis cfg $ \redis -> do
s3 <- newS3 cfg
(migrate db schemaPath productCfg)
withRedis redisCfg $ \redis -> do
s3 <- newS3 s3Cfg
let env = newEnv cfg db redis s3
withLinkedWorker (runRecovery env) $ do
let settings =
setPort (cfgListenPort cfg) $
setHost (readHost (cfgListenHost cfg)) defaultSettings
setPort (cfgListenPort httpCfg) $
setHost (readHost (cfgListenHost httpCfg)) defaultSettings
putStrLn $
"nuthatch listening on "
<> cfgListenHost cfg
<> cfgListenHost httpCfg
<> ":"
<> show (cfgListenPort cfg)
<> show (cfgListenPort httpCfg)
runTLS
(tlsSettings (cfgTlsCertPath cfg) (cfgTlsKeyPath cfg))
(tlsSettings (cfgTlsCertPath httpCfg) (cfgTlsKeyPath httpCfg))
settings
(application env)

1
nuthatch.cabal
View file

@ -113,6 +113,7 @@ library
Nuthatch.Mutation.Service
Nuthatch.Oidc
Nuthatch.Paths
Nuthatch.Product.Limits
Nuthatch.Protocol
Nuthatch.Redis
Nuthatch.Recovery

3
src/Nuthatch/App.hs
View file

@ -6,6 +6,7 @@ module Nuthatch.App
import Nuthatch.Api (api)
import Nuthatch.App.Env
import Nuthatch.Config (Config (..))
import Nuthatch.Http.Auth (authenticationHandler)
import Nuthatch.Http.Server (server, utf8Boundary)
import Servant
@ -15,5 +16,5 @@ application env =
utf8Boundary $
serveWithContext
api
(authenticationHandler (envConfig env) :. EmptyContext)
(authenticationHandler (cfgHttp (envConfig env)) :. EmptyContext)
(server env)

17
src/Nuthatch/Browse.hs
View file

@ -45,7 +45,10 @@ handleBrowseRequest env session selectedRoot selectedPrefix search continuation
let user = sessionUser session
root = maybe Personal id selectedRoot
prefix = maybe "" browsePrefixText selectedPrefix
liftIO (ensurePersonalPool (envDb env) (envConfig env) (suUserId user))
liftIO (ensurePersonalPool
(envDb env)
(cfgProduct (envConfig env))
(suUserId user))
result <- liftIO $ case search of
Nothing -> do
listed <- listEntries env user root prefix
@ -83,7 +86,7 @@ handleDeletePlanRequest env session root path continuation = do
objectPrefix
Nothing
(unPaginationToken <$> continuation)
(cfgDeletePageSize (envConfig env))
(cfgDeletePageSize (cfgProduct (envConfig env)))
page <- either
(throwOperationError status502Error
. DeletePrefixMutationRejected
@ -121,7 +124,7 @@ listEntries env user root prefix continuation = do
<> ":"
<> prefix
<> ":"
<> T.pack (show (cfgBrowsePageSize (envConfig env)))
<> T.pack (show (cfgBrowsePageSize (cfgProduct (envConfig env))))
<> ":"
<> maybe "unavailable" TE.decodeUtf8 generation
s3Prefix =
@ -143,7 +146,7 @@ listEntries env user root prefix continuation = do
s3Prefix
(Just "/")
continuation
(cfgBrowsePageSize (envConfig env))
(cfgBrowsePageSize (cfgProduct (envConfig env)))
case page of
Left err -> pure (Left (renderS3Error err))
Right listed -> do
@ -181,7 +184,7 @@ listEntries env user root prefix continuation = do
cacheSet
(envRedis env)
cacheKey
(cfgListCacheTtlSec (envConfig env))
(cfgListCacheTtlSec (cfgProduct (envConfig env)))
(BL.toStrict (A.encode result))
pure (Right result)
@ -199,11 +202,11 @@ searchEntries env user root query = do
needle = T.toCaseFold query
started <- getMonotonicTimeNSec
let deadline =
started + fromInteger (cfgSearchTimeoutMs (envConfig env) * 1000000)
started + fromInteger (cfgSearchTimeoutMs (cfgProduct (envConfig env)) * 1000000)
keysResult <- collectKeysBounded
env
s3Prefix
(cfgSearchMaxKeys (envConfig env))
(cfgSearchMaxKeys (cfgProduct (envConfig env)))
deadline
pure $ case keysResult of
Left err -> Left err

108
src/Nuthatch/Config.hs
View file

@ -1,5 +1,13 @@
module Nuthatch.Config
( Config (..)
, RuntimeConfig (..)
, HttpConfig (..)
, OidcConfig (..)
, PostgresConfig (..)
, RedisConfig (..)
, S3Config (..)
, ProductConfig (..)
, RecoveryConfig (..)
, Secret
, secret
, revealSecret
@ -17,7 +25,7 @@ import Data.Text (Text)
import qualified Data.Text as T
import qualified Data.Text.Encoding as TE
import qualified Env
import Nuthatch.Streaming (maximumMultipartBytes)
import Nuthatch.Product.Limits (maximumObjectBytes)
import qualified Text.URI as URI
newtype Secret value = Secret value
@ -29,6 +37,22 @@ revealSecret :: Secret value -> value
revealSecret (Secret value) = value
data Config = Config
{ cfgRuntime :: RuntimeConfig
, cfgHttp :: HttpConfig
, cfgOidc :: OidcConfig
, cfgPostgres :: PostgresConfig
, cfgRedis :: RedisConfig
, cfgS3 :: S3Config
, cfgProduct :: ProductConfig
, cfgRecovery :: RecoveryConfig
}
data RuntimeConfig = RuntimeConfig
{ cfgStartupAttempts :: Int
, cfgStartupDelayMs :: Int
}
data HttpConfig = HttpConfig
{ cfgListenHost :: String
, cfgListenPort :: Int
, cfgTlsCertPath :: FilePath
@ -39,41 +63,59 @@ data Config = Config
, cfgCookieName :: Text
, cfgCsrfCookieName :: Text
, cfgOidcTransactionCookieName :: Text
, cfgOidcIssuer :: Text
}
data OidcConfig = OidcConfig
{ cfgOidcIssuer :: Text
, cfgOidcClientId :: Text
, cfgOidcAdditionalAudiences :: [Text]
, cfgOidcClientSecret :: Secret Text
, cfgOidcRedirectUri :: Text
, cfgOidcTransactionMinutes :: Integer
, cfgOidcMaxIdTokenSeconds :: Integer
, cfgPgConn :: Secret String
, cfgRedisHost :: String
, cfgOidcTransactionSecret :: Secret Text
}
data PostgresConfig = PostgresConfig
{ cfgPgConn :: Secret String
}
data RedisConfig = RedisConfig
{ cfgRedisHost :: String
, cfgRedisPort :: Int
, cfgRedisDb :: Integer
, cfgS3Endpoint :: Text
, cfgThrottleBytesPerSec :: Integer
, cfgThrottleBurstBytes :: Integer
}
data S3Config = S3Config
{ cfgS3Endpoint :: Text
, cfgS3Region :: Text
, cfgS3Bucket :: Text
, cfgS3AccessKey :: Secret Text
, cfgS3SecretKey :: Secret Text
, cfgS3PathStyle :: Bool
, cfgPersonalQuotaBytes :: Integer
, cfgS3RequestTimeoutSec :: Integer
}
data ProductConfig = ProductConfig
{ cfgPersonalQuotaBytes :: Integer
, cfgSharedQuotaBytes :: Integer
, cfgMaxObjectBytes :: Integer
, cfgThrottleBytesPerSec :: Integer
, cfgThrottleBurstBytes :: Integer
, cfgListCacheTtlSec :: Integer
, cfgBrowsePageSize :: Int
, cfgSearchMaxKeys :: Int
, cfgSearchTimeoutMs :: Integer
, cfgDeletePageSize :: Int
, cfgRecoveryGraceSec :: Integer
, cfgIdempotencyWindowSec :: Integer
, cfgObjectRequestTimeoutSec :: Integer
, cfgStartupAttempts :: Int
, cfgStartupDelayMs :: Int
, cfgAuditObjectKeys :: Bool
}
data RecoveryConfig = RecoveryConfig
{ cfgRecoveryGraceSec :: Integer
, cfgObjectRequestTimeoutSec :: Integer
}
loadConfig :: IO Config
loadConfig = do
listen <- envString "NUTHATCH_LISTEN" "127.0.0.1:8080"
@ -131,7 +173,7 @@ loadConfig = do
maxObject <- envInteger "NUTHATCH_MAX_OBJECT_BYTES" (100 * 1024 * 1024)
throttleRate <- envInteger "NUTHATCH_THROTTLE_BYTES_PER_SEC" (10 * 1024 * 1024)
throttleBurst <- envInteger "NUTHATCH_THROTTLE_BURST_BYTES" (20 * 1024 * 1024)
when (maxObject <= 0 || maxObject > maximumMultipartBytes) $
when (maxObject <= 0 || maxObject > maximumObjectBytes) $
fail "NUTHATCH_MAX_OBJECT_BYTES exceeds the supported streaming range"
when (throttleRate <= 0 || throttleRate > luaExactIntegerMax) $
fail "NUTHATCH_THROTTLE_BYTES_PER_SEC exceeds the exact throttle range"
@ -167,6 +209,11 @@ loadConfig = do
when (searchTimeoutMs <= 0 || searchTimeoutMs > 60000) $
fail "NUTHATCH_SEARCH_TIMEOUT_MS must be between 1 and 60000"
pure Config
{ cfgRuntime = RuntimeConfig
{ cfgStartupAttempts = startupAttempts
, cfgStartupDelayMs = startupDelayMs
}
, cfgHttp = HttpConfig
{ cfgListenHost = host
, cfgListenPort = port
, cfgTlsCertPath = tlsCertPath
@ -177,40 +224,53 @@ loadConfig = do
, cfgCookieName = cookieName
, cfgCsrfCookieName = csrfCookie
, cfgOidcTransactionCookieName = transactionCookie
, cfgOidcIssuer = oidcIssuer
}
, cfgOidc = OidcConfig
{ cfgOidcIssuer = oidcIssuer
, cfgOidcClientId = oidcClientId
, cfgOidcAdditionalAudiences = oidcAdditionalAudiences
, cfgOidcClientSecret = oidcClientSecret
, cfgOidcRedirectUri = oidcRedirectUri
, cfgOidcTransactionMinutes = oidcTransactionMinutes
, cfgOidcMaxIdTokenSeconds = oidcMaxIdTokenSeconds
, cfgPgConn = pgConn
, cfgRedisHost = redisHost
, cfgOidcTransactionSecret = jwtSecret
}
, cfgPostgres = PostgresConfig
{ cfgPgConn = pgConn
}
, cfgRedis = RedisConfig
{ cfgRedisHost = redisHost
, cfgRedisPort = redisPort
, cfgRedisDb = redisDb
, cfgS3Endpoint = s3Endpoint
, cfgThrottleBytesPerSec = throttleRate
, cfgThrottleBurstBytes = throttleBurst
}
, cfgS3 = S3Config
{ cfgS3Endpoint = s3Endpoint
, cfgS3Region = s3Region
, cfgS3Bucket = s3Bucket
, cfgS3AccessKey = s3Access
, cfgS3SecretKey = s3Secret
, cfgS3PathStyle = pathStyle
, cfgPersonalQuotaBytes = personalQuota
, cfgS3RequestTimeoutSec = objectRequestTimeout
}
, cfgProduct = ProductConfig
{ cfgPersonalQuotaBytes = personalQuota
, cfgSharedQuotaBytes = sharedQuota
, cfgMaxObjectBytes = maxObject
, cfgThrottleBytesPerSec = throttleRate
, cfgThrottleBurstBytes = throttleBurst
, cfgListCacheTtlSec = listTtl
, cfgBrowsePageSize = browsePageSize
, cfgSearchMaxKeys = searchMax
, cfgSearchTimeoutMs = searchTimeoutMs
, cfgDeletePageSize = deletePageSize
, cfgRecoveryGraceSec = recoveryGrace
, cfgIdempotencyWindowSec = idempotencyWindow
, cfgObjectRequestTimeoutSec = objectRequestTimeout
, cfgStartupAttempts = startupAttempts
, cfgStartupDelayMs = startupDelayMs
, cfgAuditObjectKeys = auditObjectKeys
}
, cfgRecovery = RecoveryConfig
{ cfgRecoveryGraceSec = recoveryGrace
, cfgObjectRequestTimeoutSec = objectRequestTimeout
}
}
luaExactIntegerMax :: Integer
luaExactIntegerMax = 9007199254740991

12
src/Nuthatch/Db.hs
View file

@ -37,12 +37,16 @@ import Database.PostgreSQL.Simple.Migration
, runMigration
)
import qualified Database.PostgreSQL.Simple.Transaction as PGT
import Nuthatch.Config (Config (..), revealSecret)
import Nuthatch.Config
( PostgresConfig (..)
, ProductConfig (..)
, revealSecret
)
import Nuthatch.Types
type Db = Pool Connection
withDb :: Config -> (Db -> IO a) -> IO a
withDb :: PostgresConfig -> (Db -> IO a) -> IO a
withDb cfg action = do
let mk = connectPostgreSQL (BS8.pack (revealSecret (cfgPgConn cfg)))
poolCfg =
@ -50,7 +54,7 @@ withDb cfg action = do
defaultPoolConfig mk close 30 10
bracket (newPool poolCfg) destroyAllResources action
migrate :: Db -> FilePath -> Config -> IO ()
migrate :: Db -> FilePath -> ProductConfig -> IO ()
migrate db migrationDirectory cfg = withResource db $ \conn -> do
void
(PG.query_ conn
@ -72,7 +76,7 @@ migrate db migrationDirectory cfg = withResource db $ \conn -> do
configurePoolLimitConnection conn sharedPool (cfgSharedQuotaBytes cfg)
MigrationError name -> throwIO (userError ("migration failed: " <> name))
ensurePersonalPool :: Db -> Config -> UserId -> IO ()
ensurePersonalPool :: Db -> ProductConfig -> UserId -> IO ()
ensurePersonalPool db cfg uid =
configurePoolLimit db (personalPool uid) (cfgPersonalQuotaBytes cfg)

30
src/Nuthatch/Http/Auth.hs
View file

@ -47,7 +47,7 @@ import Web.Cookie
, setCookieValue
)
authenticationHandler :: Config -> AuthHandler Request SessionContext
authenticationHandler :: HttpConfig -> AuthHandler Request SessionContext
authenticationHandler cfg = mkAuthHandler $ \request -> do
let cookies = requestCookies request
sessionName = TE.encodeUtf8 (cfgCookieName cfg)
@ -90,8 +90,10 @@ authenticationHandler cfg = mkAuthHandler $ \request -> do
handleLoginRequest :: Env -> Handler LoginStart
handleLoginRequest env = do
let cfg = envConfig env
authorizationResult <- liftIO (beginAuthorization cfg)
let config = envConfig env
httpCfg = cfgHttp config
oidcCfg = cfgOidc config
authorizationResult <- liftIO (beginAuthorization oidcCfg)
authorization <- case authorizationResult of
Left _ -> do
liftIO $ audit "login_start" "rejected" Nothing
@ -102,7 +104,7 @@ handleLoginRequest env = do
pure $
addHeader
(cookieValue
(cfgOidcTransactionCookieName cfg)
(cfgOidcTransactionCookieName httpCfg)
(authorizationTransaction authorization)
"/oidc/callback"
True) $
@ -115,26 +117,30 @@ handleOidcCallback
-> Text
-> Handler LoginSuccess
handleOidcCallback env code state cookieHeader = do
let cfg = envConfig env
let config = envConfig env
httpCfg = cfgHttp config
oidcCfg = cfgOidc config
productCfg = cfgProduct config
cookies = parseCookies (TE.encodeUtf8 cookieHeader)
transactionName = TE.encodeUtf8 (cfgOidcTransactionCookieName cfg)
transactionName = TE.encodeUtf8 (cfgOidcTransactionCookieName httpCfg)
transactionBytes <- case lookup transactionName cookies of
Nothing -> rejectCallback "missing_transaction"
Just value -> pure value
transaction <- case TE.decodeUtf8' transactionBytes of
Right value | not (T.null value) && T.length value <= 8192 -> pure value
_ -> rejectCallback "malformed_transaction"
identityResult <- liftIO (redeemAuthorizationCode cfg transaction state code)
identityResult <- liftIO
(redeemAuthorizationCode oidcCfg transaction state code)
identity <- case identityResult of
Left reason -> do
liftIO $ audit "login" "rejected" Nothing [("reason", A.String reason)]
throwOperationError err400 OidcCallbackRejected
Right accepted -> pure accepted
let uid = encodeUserId (identityIssuer identity) (identitySubject identity)
liftIO (ensurePersonalPool (envDb env) cfg uid)
liftIO (ensurePersonalPool (envDb env) productCfg uid)
tokenResult <- liftIO $
mintSession
cfg
httpCfg
(identityIssuer identity)
(identitySubject identity)
(identityDisplay identity)
@ -148,14 +154,14 @@ handleOidcCallback env code state cookieHeader = do
tokenResult
csrf <- liftIO newCsrfToken
liftIO $ audit "login" "ok" (Just (unUserId uid)) []
pure (loginResponse cfg token csrf)
pure (loginResponse httpCfg token csrf)
where
rejectCallback reason = do
liftIO $ audit "login" "rejected" Nothing
[("reason", A.String reason)]
throwOperationError err400 OidcCallbackRejected
loginResponse :: Config -> SessionToken -> CsrfToken -> LoginSuccess
loginResponse :: HttpConfig -> SessionToken -> CsrfToken -> LoginSuccess
loginResponse cfg token csrf =
addHeader (cookieValue (cfgCookieName cfg) (unSessionToken token) "/" True) $
addHeader (cookieValue (cfgCsrfCookieName cfg) (unCsrfToken csrf) "/" False) $
@ -172,7 +178,7 @@ handleLogoutRequest
-> Handler LogoutSuccess
handleLogoutRequest env session providedCsrf = do
requireCsrf "logout" LogoutCsrfRejected session providedCsrf
let cfg = envConfig env
let cfg = cfgHttp (envConfig env)
liftIO $ audit "logout" "ok"
(Just (unUserId (suUserId (sessionUser session)))) []
pure $

12
src/Nuthatch/Http/Server.hs
View file

@ -168,7 +168,7 @@ handleDeleteRequest env session request = do
([ ("root", A.toJSON (rootLabel root))
, ("operation_id", A.toJSON (unOperationId operation))
, ("reason", A.String "database_unavailable")
] <> objectKeyAudit (envConfig env) (deletePath request))
] <> objectKeyAudit (cfgProduct (envConfig env)) (deletePath request))
throwOperationError err503
(DeleteMutationRejected "database unavailable")
Right value -> pure value
@ -179,7 +179,7 @@ handleDeleteRequest env session request = do
([ ("root", A.toJSON (rootLabel root))
, ("operation_id", A.toJSON (unOperationId operation))
, ("reason", A.String "object_store_unavailable")
] <> objectKeyAudit (envConfig env) (deletePath request))
] <> objectKeyAudit (cfgProduct (envConfig env)) (deletePath request))
throwOperationError err409 (DeleteMutationRejected reason)
Right value -> pure value
liftIO $ audit "delete" (mutationAuditResult result)
@ -187,12 +187,12 @@ handleDeleteRequest env session request = do
([ ("root", A.toJSON (rootLabel root))
, ("operation_id", A.toJSON (unOperationId operation))
] <> mutationReasonFields result
<> objectKeyAudit (envConfig env) (deletePath request))
<> objectKeyAudit (cfgProduct (envConfig env)) (deletePath request))
pure DeleteResponse
{ deleteResponseOperationId = operation
, deleteResponseResult = result
, deleteResponseIdempotencyWindowSeconds =
cfgIdempotencyWindowSec (envConfig env)
cfgIdempotencyWindowSec (cfgProduct (envConfig env))
}
handleDeleteBatchRequest
@ -207,7 +207,7 @@ handleDeleteBatchRequest env session request = do
session
(deleteBatchCsrf request)
let items = deleteBatchItems request
when (null items || length items > cfgDeletePageSize (envConfig env)) $ do
when (null items || length items > cfgDeletePageSize (cfgProduct (envConfig env))) $ do
liftIO $ audit "delete_batch" "rejected"
(Just (unUserId (suUserId (sessionUser session))))
[ ("root", A.toJSON (rootLabel (deleteBatchRoot request)))
@ -239,7 +239,7 @@ handleDeleteBatchRequest env session request = do
(const [("reason", A.String failureReason)])
mutationReasonFields
outcome
<> objectKeyAudit (envConfig env) (deleteBatchPath item))
<> objectKeyAudit (cfgProduct (envConfig env)) (deleteBatchPath item))
pure $ case outcome of
Left message -> DeleteItemResponse
{ deleteItemPath = deleteBatchPath item

2
src/Nuthatch/Mutation/Effects.hs
View file

@ -32,7 +32,7 @@ mutationReasonFields MutationRejected =
mutationReasonFields MutationConflict =
[("reason", A.String "operation_conflict")]
objectKeyAudit :: Config -> RelativePath -> [(Text, A.Value)]
objectKeyAudit :: ProductConfig -> RelativePath -> [(Text, A.Value)]
objectKeyAudit cfg =
objectKeyFields (cfgAuditObjectKeys cfg) . unRelativePath

4
src/Nuthatch/Mutation/Recovery.hs
View file

@ -34,7 +34,7 @@ runRecovery env = forever $ do
recoverOnce :: Env -> IO ()
recoverOnce env = do
let grace = cfgRecoveryGraceSec (envConfig env)
let grace = cfgRecoveryGraceSec (cfgRecovery (envConfig env))
batch <- withTransaction (envDb env) $ \conn -> claimActiveBatch conn 20 grace
mapConcurrently_ (recoverOne env) batch
@ -69,7 +69,7 @@ finalizeRecovered env record = do
, ("operation_id", A.toJSON (unOperationId (mrOperationId record)))
] <> mutationReasonFields mutationResult
<> objectKeyAudit
(envConfig env)
(cfgProduct (envConfig env))
(mrRelativePath record))
when (result == Just MutationSucceeded) $
invalidateMutationCaches

24
src/Nuthatch/Mutation/Service.hs
View file

@ -45,7 +45,10 @@ runPut
-> (ObjectKey -> IO (Either PutFailure ()))
-> IO (Either PutFailure MutationResult)
runPut env user root rel op size sha dispatchPut = do
ensurePersonalPool (envDb env) (envConfig env) (suUserId user)
ensurePersonalPool
(envDb env)
(cfgProduct (envConfig env))
(suUserId user)
let uid = suUserId user
key = buildObjectKey root uid rel
pool = case root of
@ -53,7 +56,7 @@ runPut env user root rel op size sha dispatchPut = do
Shared -> sharedPool
admitted <- withTransaction (envDb env) $ \conn -> do
existing <- mutationEpochExists
conn uid Put op (cfgIdempotencyWindowSec (envConfig env))
conn uid Put op (cfgIdempotencyWindowSec (cfgProduct (envConfig env)))
available <-
if existing then pure True else objectStoreAvailable (envS3 env)
if available
@ -67,9 +70,9 @@ runPut env user root rel op size sha dispatchPut = do
MutationFingerprint Put key size (Just sha)
}
AdmissionLimits
{ admissionMaxObjectBytes = cfgMaxObjectBytes (envConfig env)
{ admissionMaxObjectBytes = cfgMaxObjectBytes (cfgProduct (envConfig env))
, admissionIdempotencyWindowSec =
cfgIdempotencyWindowSec (envConfig env)
cfgIdempotencyWindowSec (cfgProduct (envConfig env))
}
else pure Nothing
case admitted of
@ -102,7 +105,10 @@ runDelete
-> OperationId
-> IO (Either Text MutationResult)
runDelete env user root rel op = do
ensurePersonalPool (envDb env) (envConfig env) (suUserId user)
ensurePersonalPool
(envDb env)
(cfgProduct (envConfig env))
(suUserId user)
let uid = suUserId user
key = buildObjectKey root uid rel
pool = case root of
@ -110,7 +116,7 @@ runDelete env user root rel op = do
Shared -> sharedPool
admitted <- withTransaction (envDb env) $ \conn -> do
existing <- mutationEpochExists
conn uid Delete op (cfgIdempotencyWindowSec (envConfig env))
conn uid Delete op (cfgIdempotencyWindowSec (cfgProduct (envConfig env)))
available <-
if existing then pure True else objectStoreAvailable (envS3 env)
if available
@ -124,9 +130,9 @@ runDelete env user root rel op = do
MutationFingerprint Delete key 0 Nothing
}
AdmissionLimits
{ admissionMaxObjectBytes = cfgMaxObjectBytes (envConfig env)
{ admissionMaxObjectBytes = cfgMaxObjectBytes (cfgProduct (envConfig env))
, admissionIdempotencyWindowSec =
cfgIdempotencyWindowSec (envConfig env)
cfgIdempotencyWindowSec (cfgProduct (envConfig env))
}
else pure Nothing
case admitted of
@ -156,7 +162,7 @@ dispatchAndFinalize
-> IO (Either failure ())
-> IO (Either failure MutationResult)
dispatchAndFinalize env user root record timeoutFailure dispatch = do
let cfg = envConfig env
let cfg = cfgRecovery (envConfig env)
requestTimeoutSec = cfgObjectRequestTimeoutSec cfg
prepared <- withTransaction (envDb env) $ \conn -> do
moved <- markApplying conn record

22
src/Nuthatch/Oidc.hs
View file

@ -66,7 +66,7 @@ import Network.HTTP.Simple
, setRequestMethod
)
import Network.HTTP.Types.URI (renderQuery)
import Nuthatch.Config (Config (..), revealSecret)
import Nuthatch.Config (OidcConfig (..), revealSecret)
import System.Entropy (getEntropy)
data AuthorizationRequest = AuthorizationRequest
@ -186,7 +186,7 @@ instance A.ToJSON IdTokenClaims where
, ("preferred_username", maybe A.Null A.String (idTokenPreferred claims))
]
beginAuthorization :: Config -> IO (Either Text AuthorizationRequest)
beginAuthorization :: OidcConfig -> IO (Either Text AuthorizationRequest)
beginAuthorization cfg = do
discoveryResult <- fetchDiscovery cfg
case discoveryResult of
@ -220,7 +220,7 @@ beginAuthorization cfg = do
}
redeemAuthorizationCode
:: Config
:: OidcConfig
-> Text
-> Text
-> Text
@ -255,7 +255,7 @@ redeemAuthorizationCode cfg transactionToken returnedState code = do
(transactionNonce transaction)
idToken
fetchDiscovery :: Config -> IO (Either Text Discovery)
fetchDiscovery :: OidcConfig -> IO (Either Text Discovery)
fetchDiscovery cfg = catchOidcHttp "OIDC discovery request failed" $ do
request <- parseRequest $
T.unpack (cfgOidcIssuer cfg <> "/.well-known/openid-configuration")
@ -270,7 +270,7 @@ fetchDiscovery cfg = catchOidcHttp "OIDC discovery request failed" $ do
| otherwise -> Left "OIDC discovery issuer mismatch"
exchangeCode
:: Config
:: OidcConfig
-> Discovery
-> TransactionClaims
-> Text
@ -299,7 +299,7 @@ exchangeCode cfg discovery transaction code = do
Right tokenResponse -> Right (tokenResponseIdToken tokenResponse)
fetchAndVerifyIdToken
:: Config
:: OidcConfig
-> Discovery
-> Text
-> Text
@ -317,7 +317,7 @@ fetchAndVerifyIdToken cfg discovery expectedNonce compact = do
verifyIdToken cfg jwks now expectedNonce compact
verifyIdToken
:: Config
:: OidcConfig
-> JWKSet
-> UTCTime
-> Text
@ -384,7 +384,7 @@ verifyIdToken cfg jwks now expectedNonce compact = do
}
mintTransaction
:: Config
:: OidcConfig
-> Text
-> Text
-> Text
@ -410,16 +410,16 @@ mintTransaction cfg state nonce verifier = do
, transactionRedirectUri = cfgOidcRedirectUri cfg
, transactionReturnTarget = "/"
}
key = fromOctets (TE.encodeUtf8 (revealSecret (cfgJwtSecret cfg)))
key = fromOctets (TE.encodeUtf8 (revealSecret (cfgOidcTransactionSecret cfg)))
result <- runJOSE $
encodeCompact <$> signJWT key (newJWSHeaderProtected HS256) claims
pure $ case result of
Left (_ :: JWTError) -> Left "OIDC transaction creation failed"
Right token -> Right (TE.decodeUtf8 (BL.toStrict token))
verifyTransaction :: Config -> Text -> IO (Either Text TransactionClaims)
verifyTransaction :: OidcConfig -> Text -> IO (Either Text TransactionClaims)
verifyTransaction cfg compact = do
let key = fromOctets (TE.encodeUtf8 (revealSecret (cfgJwtSecret cfg)))
let key = fromOctets (TE.encodeUtf8 (revealSecret (cfgOidcTransactionSecret cfg)))
settings =
defaultJWTValidationSettings (== "nuthatch-oidc-callback")
& issuerPredicate .~ (== "nuthatch")

10
src/Nuthatch/Product/Limits.hs Normal file
View file

@ -0,0 +1,10 @@
module Nuthatch.Product.Limits
( multipartPartBytes
, maximumObjectBytes
) where
multipartPartBytes :: Int
multipartPartBytes = 5 * 1024 * 1024
maximumObjectBytes :: Integer
maximumObjectBytes = toInteger multipartPartBytes * 10000

10
src/Nuthatch/Redis.hs
View file

@ -28,13 +28,13 @@ import Database.Redis
, setex
)
import qualified Database.Redis as R
import Nuthatch.Config (Config (..))
import Nuthatch.Config (RedisConfig (..))
import Nuthatch.Recovery (trySynchronous)
import Nuthatch.Types
type Redis = Connection
withRedis :: Config -> (Redis -> IO a) -> IO a
withRedis :: RedisConfig -> (Redis -> IO a) -> IO a
withRedis cfg action = do
conn <- connect defaultConnectInfo
{ connectHost = cfgRedisHost cfg
@ -93,16 +93,16 @@ cacheGenerationKey uid root =
Personal -> unUserId uid <> ":personal"
Shared -> "shared"
throttleCheck :: Redis -> Config -> UserId -> Integer -> IO (Either Text Bool)
throttleCheck :: Redis -> RedisConfig -> UserId -> Integer -> IO (Either Text Bool)
throttleCheck conn cfg uid bytes =
throttleRequest conn cfg uid bytes False
throttleTake :: Redis -> Config -> UserId -> Integer -> IO (Either Text Bool)
throttleTake :: Redis -> RedisConfig -> UserId -> Integer -> IO (Either Text Bool)
throttleTake conn cfg uid bytes = throttleRequest conn cfg uid bytes True
throttleRequest
:: Redis
-> Config
-> RedisConfig
-> UserId
-> Integer
-> Bool

6
src/Nuthatch/S3.hs
View file

@ -61,7 +61,7 @@ import qualified Network.HTTP.Client as HTTP
import Network.HTTP.Client.TLS (tlsManagerSettings)
import Network.HTTP.Types.Status (statusCode)
import Network.URI (URI (..), URIAuth (..), parseURI)
import Nuthatch.Config (Config (..), revealSecret)
import Nuthatch.Config (S3Config (..), revealSecret)
import Nuthatch.Streaming
( BandwidthError (..)
, UploadStreamError (..)
@ -112,7 +112,7 @@ data ListPage = ListPage
}
deriving stock (Eq, Show)
newS3 :: Config -> IO S3
newS3 :: S3Config -> IO S3
newS3 cfg = do
(protocol, endpointHost, endpointPort) <- parseEndpoint (cfgS3Endpoint cfg)
credentials <- Aws.makeCredentials
@ -122,7 +122,7 @@ newS3 cfg = do
AwsCore.HTTP -> 80
AwsCore.HTTPS -> 443
requestTimeoutMicros =
fromInteger (cfgObjectRequestTimeoutSec cfg * 1000000)
fromInteger (cfgS3RequestTimeoutSec cfg * 1000000)
signedHost =
if endpointPort == defaultPort
then endpointHost

6
src/Nuthatch/Session.hs
View file

@ -45,7 +45,7 @@ import Data.Text (Text)
import qualified Data.Text as T
import qualified Data.Text.Encoding as TE
import Data.Time.Clock (addUTCTime, getCurrentTime)
import Nuthatch.Config (Config (..), revealSecret)
import Nuthatch.Config (HttpConfig (..), revealSecret)
import Nuthatch.Paths (encodeUserId)
import Nuthatch.Types
import System.Entropy (getEntropy)
@ -87,7 +87,7 @@ instance A.ToJSON SessionClaims where
(displayName claims)
value -> value
mintSession :: Config -> Text -> Text -> Maybe Text -> IO (Either Text SessionToken)
mintSession :: HttpConfig -> Text -> Text -> Maybe Text -> IO (Either Text SessionToken)
mintSession cfg issuer subject display = do
now <- getCurrentTime
let expires = addUTCTime (fromInteger (cfgSessionHours cfg * 3600)) now
@ -116,7 +116,7 @@ mintSession cfg issuer subject display = do
Right compact ->
validateSessionToken (TE.decodeUtf8 (BL.toStrict compact))
verifySession :: Config -> SessionToken -> IO (Either Text SessionUser)
verifySession :: HttpConfig -> SessionToken -> IO (Either Text SessionUser)
verifySession cfg (SessionToken token) = do
let applicationIssuer = "nuthatch"
applicationAudience = "nuthatch"

9
src/Nuthatch/Streaming.hs
View file

@ -13,6 +13,10 @@ import qualified Data.ByteString as BS
import qualified Data.ByteString.Base16 as B16
import Data.IORef
import qualified Data.Text.Encoding as TE
import Nuthatch.Product.Limits
( maximumObjectBytes
, multipartPartBytes
)
import Nuthatch.Types
import Servant.Types.SourceT
( SourceT
@ -34,11 +38,8 @@ data BandwidthError
instance Exception BandwidthError
multipartPartBytes :: Int
multipartPartBytes = 5 * 1024 * 1024
maximumMultipartBytes :: Integer
maximumMultipartBytes = toInteger multipartPartBytes * 10000
maximumMultipartBytes = maximumObjectBytes
data UploadState = UploadState
{ usPending :: [BS.ByteString]

24
src/Nuthatch/Transfer.hs
View file

@ -66,7 +66,7 @@ handleDownloadRequest env session root path requestedRange = do
liftIO $ audit "download" "ok" (Just (unUserId (suUserId user)))
([ ("root", A.toJSON (rootLabel root))
, ("bytes", A.toJSON (oiSize info))
] <> objectKeyAudit (envConfig env) path)
] <> objectKeyAudit (cfgProduct (envConfig env)) path)
pure $
addHeader (contentDisposition (basename path)) $
addHeader "nosniff" $
@ -97,14 +97,14 @@ handleUploadRequest
requireCsrf "put" UploadCsrfRejected session providedCsrf
let user = sessionUser session
size = unDeclaredSize declaredSize
configuredMaximum = cfgMaxObjectBytes (envConfig env)
configuredMaximum = cfgMaxObjectBytes (cfgProduct (envConfig env))
when (size > configuredMaximum) $ do
liftIO $ audit "put" "rejected" (Just (unUserId (suUserId user)))
([ ("root", A.toJSON (rootLabel root))
, ("bytes", A.toJSON size)
, ("operation_id", A.toJSON (unOperationId operation))
, ("reason", A.String "object_too_large")
] <> objectKeyAudit (envConfig env) path)
] <> objectKeyAudit (cfgProduct (envConfig env)) path)
throwOperationError err413 UploadTooLarge
requireUploadBandwidth env user root path operation size
putAttempt <- liftIO $
@ -134,7 +134,7 @@ handleUploadRequest
, ("bytes", A.toJSON size)
, ("operation_id", A.toJSON (unOperationId operation))
, ("reason", A.String "database_unavailable")
] <> objectKeyAudit (envConfig env) path)
] <> objectKeyAudit (cfgProduct (envConfig env)) path)
throwOperationError err503
(UploadMutationRejected "database unavailable")
Right value -> pure value
@ -145,7 +145,7 @@ handleUploadRequest
, ("bytes", A.toJSON size)
, ("operation_id", A.toJSON (unOperationId operation))
, ("reason", A.String (putFailureReason failure))
] <> objectKeyAudit (envConfig env) path)
] <> objectKeyAudit (cfgProduct (envConfig env)) path)
throwPutFailure failure
Right value -> pure value
liftIO $ audit "put" (mutationAuditResult result)
@ -154,12 +154,12 @@ handleUploadRequest
, ("bytes", A.toJSON size)
, ("operation_id", A.toJSON (unOperationId operation))
] <> mutationReasonFields result
<> objectKeyAudit (envConfig env) path)
<> objectKeyAudit (cfgProduct (envConfig env)) path)
pure UploadResponse
{ uploadResponseOperationId = operation
, uploadResponseResult = result
, uploadResponseIdempotencyWindowSeconds =
cfgIdempotencyWindowSec (envConfig env)
cfgIdempotencyWindowSec (cfgProduct (envConfig env))
}
requireDownloadBandwidth :: Env -> SessionUser -> Integer -> Handler ()
@ -167,7 +167,7 @@ requireDownloadBandwidth env user size = do
available <- liftIO $
throttleCheck
(envRedis env)
(envConfig env)
(cfgRedis (envConfig env))
(suUserId user)
(min size (toInteger (transferPieceSize env)))
case available of
@ -187,7 +187,7 @@ requireUploadBandwidth env user root path operation size = do
available <- liftIO $
throttleCheck
(envRedis env)
(envConfig env)
(cfgRedis (envConfig env))
(suUserId user)
(min size (toInteger (transferPieceSize env)))
case available of
@ -202,14 +202,14 @@ requireUploadBandwidth env user root path operation size = do
, ("bytes", A.toJSON size)
, ("operation_id", A.toJSON (unOperationId operation))
, ("reason", A.String reason)
] <> objectKeyAudit (envConfig env) path)
] <> objectKeyAudit (cfgProduct (envConfig env)) path)
throwOperationError status response
inspectBandwidth :: Env -> SessionUser -> BS.ByteString -> IO ()
inspectBandwidth env user chunk = do
result <- throttleTake
(envRedis env)
(envConfig env)
(cfgRedis (envConfig env))
(suUserId user)
(toInteger (BS.length chunk))
case result of
@ -219,7 +219,7 @@ inspectBandwidth env user chunk = do
transferPieceSize :: Env -> Int
transferPieceSize env =
fromInteger (max 1 (min 65536 (cfgThrottleBurstBytes (envConfig env))))
fromInteger (max 1 (min 65536 (cfgThrottleBurstBytes (cfgRedis (envConfig env)))))
throwPutFailure :: PutFailure -> Handler a
throwPutFailure failure =

8
test/FixtureProbe.hs
View file

@ -8,7 +8,7 @@ import qualified Data.ByteString.Base16 as B16
import qualified Data.Text as Text
import qualified Data.Text.Encoding as TextEncoding
import Database.PostgreSQL.Simple (Only (..))
import Nuthatch.Config (loadConfig)
import Nuthatch.Config (Config (..), loadConfig)
import qualified Nuthatch.Db as Db
import Nuthatch.Probe.Assert
import Nuthatch.Probe.Curl
@ -27,14 +27,14 @@ main = do
setEnv "NUTHATCH_OIDC_CLIENT_SECRET" "synthetic-fixture-client-secret"
config <- loadConfig
verifyProfileFile profile lane pgConnection
Db.withDb config $ \database -> do
Db.withDb (cfgPostgres config) $ \database -> do
_ <- Db.execute_ database
"CREATE TEMP TABLE lane_canary (value text NOT NULL)"
_ <- Db.execute database
"INSERT INTO lane_canary VALUES (?)"
(Only (Text.pack profile))
pure ()
withRedis config $ \redis -> do
withRedis (cfgRedis config) $ \redis -> do
cacheSet redis "lane-canary" 60 (B8.pack profile)
cacheGet redis "lane-canary" >>=
assertEqual "coordination lane canary" (Just (B8.pack profile))
@ -42,7 +42,7 @@ main = do
discovery <- curl [issuer <> "/.well-known/openid-configuration"]
assertEqual "OIDC discovery status" 200 (curlStatus discovery)
assertContains "OIDC discovery issuer" "\"issuer\"" (curlBody discovery)
store <- newS3 config
store <- newS3 (cfgS3 config)
let key = ObjectKey "lane-canary"
body = BL.fromStrict (B8.pack profile)
digest =

22
test/Integration.hs
View file

@ -13,6 +13,8 @@ import qualified Data.Text.Encoding as TE
import Database.PostgreSQL.Simple (Only (..))
import Nuthatch.Config
( Config (..)
, ProductConfig (..)
, S3Config (..)
, loadConfig
, revealSecret
, secret
@ -33,13 +35,13 @@ import Nuthatch.Types
main :: IO ()
main = do
config <- loadConfig
withDb config $ \database -> do
migrate database "sql" config
migrate database "sql" config
withDb (cfgPostgres config) $ \database -> do
migrate database "sql" (cfgProduct config)
migrate database "sql" (cfgProduct config)
verifySchema database
verifyAdmission database config
verifyRowDecoder database
store <- newS3 config
store <- newS3 (cfgS3 config)
verifyStorage config store
putStrLn "Storage foundation integration gate passed"
@ -112,18 +114,19 @@ verifyAdmission database config = do
let personalUser = UserId "personal-limit-user"
personal = personalPool personalUser
productCfg = cfgProduct config
ensurePersonalPool database
config { cfgPersonalQuotaBytes = 30 }
productCfg { cfgPersonalQuotaBytes = 30 }
personalUser
ensurePersonalPool database
config { cfgPersonalQuotaBytes = 40 }
productCfg { cfgPersonalQuotaBytes = 40 }
personalUser
_ <- Db.execute database
"UPDATE pools SET reserved_bytes = 35 WHERE pool_id = ?"
(Only (unPoolId personal))
personalLower <- try
(ensurePersonalPool database
config { cfgPersonalQuotaBytes = 34 }
productCfg { cfgPersonalQuotaBytes = 34 }
personalUser)
:: IO (Either SomeException ())
assert "unsafe personal limit reduction rejected" (isLeft personalLower)
@ -326,9 +329,10 @@ verifyStorage config store = do
missing <- getObjectBytes store (ObjectKey "shared/missing")
assertEqual "typed not-found error" (Left S3NotFound) missing
deniedStore <- newS3 config
let s3Cfg = cfgS3 config
deniedStore <- newS3 s3Cfg
{ cfgS3SecretKey =
secret (revealSecret (cfgS3SecretKey config) <> "-invalid")
secret (revealSecret (cfgS3SecretKey s3Cfg) <> "-invalid")
}
denied <- listObjects deniedStore "" Nothing Nothing 1
case denied of

Some files were not shown because too many files have changed in this diff Show more