@@ -29,10 +29,16 @@ federationScheme = "https://" -- scheme for federation HTTP calls
29-- slow peer from consuming all concurrency slots, though truly hung connections still hold
29-- slow peer from consuming all concurrency slots, though truly hung connections still hold
30-- one slot until the forked task terminates.
30-- one slot until the forked task terminates.
31-- Message and data size limits
31-- Message and data size limits
32−-- TODO: If Knot adds HTTP-layer configuration, enforce a request body size limit
32+-- IMPORTANT: Knot has no HTTP-layer request body size limit. Field-level length checks
33−-- (e.g., 256KB) to prevent large-payload DoS. Currently field-level length checks
34−-- (maxBlobLen, maxPubkeyLen, etc.) validate after parsing, so a multi-GB JSON body
33+-- (maxBlobLen, maxPubkeyLen, etc.) validate after JSON parsing, so a multi-GB body is
35−-- would be fully parsed before rejection. A framework-level limit would reject early.
34+-- fully parsed before rejection. Deploy behind a reverse proxy with a body size limit:
35+-- Caddy: request_body { max_size 256KB }
36+-- nginx: client_max_body_size 256k;
37+-- HAProxy: tune.maxrewrite 256k
38+-- Without this, the server is vulnerable to large-payload DoS attacks. The maxBlobLen
39+-- limit (64KB hex = 32KB binary) bounds legitimate message payloads; snapshots and
40+-- batch requests may be larger but are authenticated via maintenance secret.
41+-- TODO: If Knot adds HTTP-layer configuration, enforce a request body size limit natively.
36messageMaxAge : Int<Ms>
42messageMaxAge : Int<Ms>
37messageMaxAge = 2592000000 -- 30 days
43messageMaxAge = 2592000000 -- 30 days
38maxPresenceTtl : Int<S>
44maxPresenceTtl : Int<S>
@@ -102,6 +108,7 @@ federationPairRateWindowMs : Int<Ms>
102federationPairRateWindowMs = 60000 -- 1 minute
108federationPairRateWindowMs = 60000 -- 1 minute
103maxFederationPairRateEntries = 5000
109maxFederationPairRateEntries = 5000
104-- Global request rate limiting
110-- Global request rate limiting
111+maxActiveRequests = 500 -- max concurrent requests before rejecting with overloaded (0 = disable)
105maxGlobalRequestRate = 1000 -- max total requests per window
112maxGlobalRequestRate = 1000 -- max total requests per window
106globalRateWindowMs : Int<Ms>
113globalRateWindowMs : Int<Ms>
107globalRateWindowMs = 1000
114globalRateWindowMs = 1000
@@ -226,6 +233,12 @@ autoSnapshotInterval : Int<Ms>
226autoSnapshotInterval = 300000 -- 5 minutes; periodic state summary to stdout (0 to disable)
233autoSnapshotInterval = 300000 -- 5 minutes; periodic state summary to stdout (0 to disable)
227-- Full snapshot auto-export interval: emit [SNAPSHOT_DATA] JSON to stdout for sidecar capture.
234-- Full snapshot auto-export interval: emit [SNAPSHOT_DATA] JSON to stdout for sidecar capture.
228-- 0 to disable. When enabled, a crash loses at most this interval of federation state.
235-- 0 to disable. When enabled, a crash loses at most this interval of federation state.
236+-- SECURITY: auto-exported snapshots contain unencrypted federation state (presence, acked
237+-- message IDs, peer failure records, forward nonces). The data does not include message
238+-- content (E2E encrypted blobs are ephemeral) but does reveal metadata: which pubkeys are
239+-- online, which servers they federate with, and delivery timing. Ensure the log pipeline
240+-- is access-controlled and encrypted at rest. Set to 0 in high-security deployments
241+-- and use manual GET /maintenance/snapshot with authenticated access instead.
229-- Only emits when there's meaningful state to persist (presence/forwards/ackedIds > 0).
242-- Only emits when there's meaningful state to persist (presence/forwards/ackedIds > 0).
230autoSnapshotExportInterval : Int<Ms>
243autoSnapshotExportInterval : Int<Ms>
231autoSnapshotExportInterval = 3600000 -- 1 hour; full state export to stdout (0 to disable)
244autoSnapshotExportInterval = 3600000 -- 1 hour; full state export to stdout (0 to disable)
@@ -597,6 +610,11 @@ type SenderDelivery = {senderKey: Text, messageId: MessageIdHex, toKey: PubkeyHe
597-- lastPresenceRefresh), so skipped/delayed prune cycles don't cause bucket gaps.
610-- lastPresenceRefresh), so skipped/delayed prune cycles don't cause bucket gaps.
598*presenceRefreshGroup : Int
611*presenceRefreshGroup : Int
613+-- Round-robin nonce pruning: tracks which bucket group (0-3) to scan next.
614+-- Each prune cycle scans 4 of 16 nonce buckets, cycling through all 16 over 4 cycles.
615+-- Nonces have 60-day retention so 30-second pruning granularity is unnecessary.
616+*noncePruneGroup : Int
600*challengeRates : [ChallengeRateEntry]
618*challengeRates : [ChallengeRateEntry]
602-- Separate revoke rate limiter (prevents challenge floods from exhausting revoke budget)
620-- Separate revoke rate limiter (prevents challenge floods from exhausting revoke budget)
@@ -1176,6 +1194,15 @@ warnOldProtocol = \fromServer pv ->
1176 when (pv > 0 && pv < federationProtocolVersion)
1194 when (pv > 0 && pv < federationProtocolVersion)
1177 (logWarn ("federation: peer " ++ fromServer ++ " using protocol v" ++ show pv ++ " (current: v" ++ show federationProtocolVersion ++ ")"))
1195 (logWarn ("federation: peer " ++ fromServer ++ " using protocol v" ++ show pv ++ " (current: v" ++ show federationProtocolVersion ++ ")"))
1197+-- Log per-server detail for stale in-flight entries (hung connection debugging).
1198+-- Emits one structured event per server with slot count and duration.
1199+logStaleInFlight = \entries t ->
1200+ case headOf entries of
1201+ Nothing {} -> yield {}
1202+ Just {value: r} -> do
1203+ logEvent "error" "stale_inflight" (r.server ++ " " ++ show r.count ++ " slots hung for " ++ show (t - r.firstStartedAt) ++ "ms")
1204+ logStaleInFlight (do e <- entries; where e.server != r.server; yield e) t
1179-- Fast path: check if a string contains only safe JSON characters (hex digits,
1206-- Fast path: check if a string contains only safe JSON characters (hex digits,
1180-- alphanumerics, common punctuation). If so, no escaping is needed.
1207-- alphanumerics, common punctuation). If so, no escaping is needed.
1181-- Returns False on the first unsafe character. Tail-recursive (TCO-safe in Knot).
1208-- Returns False on the first unsafe character. Tail-recursive (TCO-safe in Knot).
@@ -1494,8 +1521,16 @@ isBadServerName = \sn ->
1494 || containsDoubleEncoding sn
1521 || containsDoubleEncoding sn
1495 -- Kubernetes / service mesh internal DNS
1522 -- Kubernetes / service mesh internal DNS
1496 || containsSuffix ".svc" sn
1523 || containsSuffix ".svc" sn
1524+ || containsSuffix ".cluster.local" sn
1497 -- HashiCorp Consul service discovery
1525 -- HashiCorp Consul service discovery
1498 || containsSuffix ".consul" sn
1526 || containsSuffix ".consul" sn
1527+ || containsSuffix ".node.consul" sn
1528+ || containsSuffix ".service.consul" sn
1529+ -- Cloud provider internal DNS (SSRF targets for metadata/internal services)
1530+ || containsSuffix ".compute.internal" sn -- GCP internal DNS
1531+ || containsSuffix ".ec2.internal" sn -- AWS EC2 internal DNS
1532+ || containsSuffix ".amazonaws.com" sn -- AWS service endpoints (S3, SQS, etc.)
1533+ || matchesHostname sn "metadata.google.internal" -- GCP metadata server
1499 -- NOTE: Unicode/IDN defense is provided by containsBadUrlChar above, which rejects
1534 -- NOTE: Unicode/IDN defense is provided by containsBadUrlChar above, which rejects
1500 -- any character not in [0-9a-z-.:] via isAllowedServerChar. This blocks homoglyph
1535 -- any character not in [0-9a-z-.:] via isAllowedServerChar. This blocks homoglyph
1501 -- attacks (e.g., Greek omicron in "lοcalhost") assuming Knot's Text type preserves
1536 -- attacks (e.g., Greek omicron in "lοcalhost") assuming Knot's Text type preserves
@@ -2476,7 +2511,7 @@ writeAllNonces = \nonces -> do
2476 writeNonceBucket 14 partitioned.b14
2511 writeNonceBucket 14 partitioned.b14
2477 writeNonceBucket 15 partitioned.b15
2512 writeNonceBucket 15 partitioned.b15
2479−-- Prune nonces across all 16 buckets
2514+-- Prune nonces across all 16 buckets (used by flush and restore)
2480pruneAllNonceBuckets : Int -> Timestamp -> IO ()
2515pruneAllNonceBuckets : Int -> Timestamp -> IO ()
2481pruneAllNonceBuckets = \b t ->
2516pruneAllNonceBuckets = \b t ->
2482 if b > 15 then yield {}
2517 if b > 15 then yield {}
@@ -2485,6 +2520,25 @@ pruneAllNonceBuckets = \b t ->
2485 writeNonceBucket b (do n <- bucket; where t - n.seenAt < forwardNonceRetention; yield n)
2520 writeNonceBucket b (do n <- bucket; where t - n.seenAt < forwardNonceRetention; yield n)
2486 pruneAllNonceBuckets (b + 1) t
2521 pruneAllNonceBuckets (b + 1) t
2523+-- Round-robin nonce pruning: scan 4 of 16 buckets per call.
2524+-- With 60-day retention and 30-second prune intervals, full coverage every ~2 minutes
2525+-- is sufficient. Reduces prune cycle cost by 75% vs scanning all 16 buckets.
2526+pruneNonceBucketsRoundRobin : Timestamp -> IO ()
2527+pruneNonceBucketsRoundRobin = \t -> do
2528+ group <- *noncePruneGroup
2529+ let startBucket = (group * 4) - ((group * 4) / 16) * 16
2530+ pruneNonceBucketRange startBucket (startBucket + 3) t
2531+ set *noncePruneGroup = if group >= 3 then 0 else group + 1
2533+pruneNonceBucketRange : Int -> Int -> Timestamp -> IO ()
2534+pruneNonceBucketRange = \b endB t ->
2535+ if b > endB || b > 15 then yield {}
2537+ bucket <- readNonceBucket b
2538+ let pruned = do n <- bucket; where t - n.seenAt < forwardNonceRetention; yield n
2539+ when (count pruned < count bucket) (writeNonceBucket b pruned)
2540+ pruneNonceBucketRange (b + 1) endB t
2488-- ============================================
2542-- ============================================
2489-- Forward Bucket Helpers
2543-- Forward Bucket Helpers
2490-- ============================================
2544-- ============================================
@@ -3262,10 +3316,16 @@ pruneExpired = \t -> do
3262 pruneAllAckedBuckets 0 retainCutoff
3316 pruneAllAckedBuckets 0 retainCutoff
3263 -- DeliveredIds: per-bucket pruning (same pattern as acked)
3317 -- DeliveredIds: per-bucket pruning (same pattern as acked)
3264 pruneAllDeliveredBuckets 0 retainCutoff
3318 pruneAllDeliveredBuckets 0 retainCutoff
3265− -- Overflow + failures: smaller collections, grouped together
3319+ -- Overflow + failures: smaller collections, grouped together.
3320+ -- Stale peer cleanup: servers still in knownPeerServers (have active presence) keep
3321+ -- their failure records for the full maxServerDowntime (24h). Servers with no presence
3322+ -- are pruned aggressively after 1h — they have no local keys to forward to and their
3323+ -- failure records just inflate circuit breaker counts.
3266 atomic do
3324 atomic do
3267 forwardFailures <- *forwardFailures
3325 forwardFailures <- *forwardFailures
3326+ peers <- *knownPeerServers
3327+ let isKnown = \server -> any peers (\p -> p == server)
3268− set *forwardFailures = do f <- forwardFailures; where t - f.failedAt < maxServerDowntime; yield f
3328+ set *forwardFailures = do f <- forwardFailures; where (if isKnown f.server then t - f.failedAt < maxServerDowntime else t - f.failedAt < sessionExpiry); yield f
3269 ackOverflow <- *ackOverflow
3329 ackOverflow <- *ackOverflow
3270 set *ackOverflow = do a <- ackOverflow; where a.ackedAt > retainCutoff; yield a
3330 set *ackOverflow = do a <- ackOverflow; where a.ackedAt > retainCutoff; yield a
3271 -- Prune rate limiters (split into smaller atomic blocks to reduce lock contention)
3331 -- Prune rate limiters (split into smaller atomic blocks to reduce lock contention)
@@ -3319,8 +3379,15 @@ pruneExpired = \t -> do
3319 inFlight <- *federationInFlight
3379 inFlight <- *federationInFlight
3320 let staleThreshold = federationFetchTimeoutMs * 5 / 4
3380 let staleThreshold = federationFetchTimeoutMs * 5 / 4
3321 let staleInFlight = do r <- inFlight; where r.count > 0 && r.firstStartedAt > 0 && t - r.firstStartedAt >= staleThreshold; yield r
3381 let staleInFlight = do r <- inFlight; where r.count > 0 && r.firstStartedAt > 0 && t - r.firstStartedAt >= staleThreshold; yield r
3322− when (count staleInFlight > 0)
3382+ when (count staleInFlight > 0) (do
3323− (logWarn ("pruneExpired: cleaning " ++ show (count staleInFlight) ++ " stale in-flight entries (leaked slots)"))
3383+ logWarn ("pruneExpired: cleaning " ++ show (count staleInFlight) ++ " stale in-flight entries (leaked slots)")
3384+ -- Per-server detail for debugging hung connections: log which peers are stuck and for how long
3385+ logStaleInFlight staleInFlight t)
3386+ -- Also warn about connections approaching the stale threshold (early detection)
3387+ let hungThreshold = federationFetchTimeoutMs
3388+ let hungInFlight = do r <- inFlight; where r.count > 0 && r.firstStartedAt > 0 && t - r.firstStartedAt >= hungThreshold && t - r.firstStartedAt < staleThreshold; yield r
3389+ when (count hungInFlight > 0)
3390+ (logEvent "warn" "hung_federation" ("connections approaching stale threshold: " ++ show (count hungInFlight) ++ " slots"))
3324 set *federationInFlight = do r <- inFlight; where r.count > 0 && (r.firstStartedAt == 0 || t - r.firstStartedAt < staleThreshold); yield r
3391 set *federationInFlight = do r <- inFlight; where r.count > 0 && (r.firstStartedAt == 0 || t - r.firstStartedAt < staleThreshold); yield r
3325 -- Prune stale federation replay dedup entries
3392 -- Prune stale federation replay dedup entries
3326 recentFedReqs <- *recentFederationRequests
3393 recentFedReqs <- *recentFederationRequests
@@ -3328,8 +3395,8 @@ pruneExpired = \t -> do
3328 -- Prune expired federation verification tokens
3395 -- Prune expired federation verification tokens
3329 pending <- *pendingVerifications
3396 pending <- *pendingVerifications
3330 set *pendingVerifications = do p <- pending; where t - p.createdAt < verificationTokenExpiry; yield p
3397 set *pendingVerifications = do p <- pending; where t - p.createdAt < verificationTokenExpiry; yield p
3331− -- Prune expired forward nonces per bucket (extended replay protection)
3398+ -- Prune expired forward nonces (round-robin: 4 of 16 buckets per cycle)
3332− pruneAllNonceBuckets 0 t
3399+ pruneNonceBucketsRoundRobin t
3333 -- Prune gossip event rate entries
3400 -- Prune gossip event rate entries
3334 gossipEventRates <- *gossipEventRates
3401 gossipEventRates <- *gossipEventRates
3335 set *gossipEventRates = do r <- gossipEventRates; where t - r.windowStart < gossipEventRateWindowMs; yield r
3402 set *gossipEventRates = do r <- gossipEventRates; where t - r.windowStart < gossipEventRateWindowMs; yield r
@@ -3430,46 +3497,73 @@ mergeGossipPresence = \presence deduped trustedOffline fromServer t ->
3430 withinCapacity: capResult.withinCapacity}
3497 withinCapacity: capResult.withinCapacity}
3500+-- Adaptive rate limiting: scale rate limits down when the server is under heavy load.
3501+-- Returns a scaling factor from 0 to 100 (percentage of full rate to allow).
3502+-- 100% at <=80% utilization, linearly decreasing to 10% at 100% utilization.
3503+-- Uses the highest utilization across messages, sessions, and presence as the
3504+-- bottleneck metric. 10% floor prevents complete lockout (admin can still act).
3505+adaptiveRateScale : Int -> Int -> Int -> Int
3506+adaptiveRateScale = \msgCount sessCount presCount ->
3507+ let msgPct = if maxMessages > 0 then msgCount * 100 / maxMessages else 0 in
3508+ let sessPct = if maxSessions > 0 then sessCount * 100 / maxSessions else 0 in
3509+ let presPct = if maxPresenceEntries > 0 then presCount * 100 / maxPresenceEntries else 0 in
3510+ let peak = max msgPct (max sessPct presPct) in
3511+ if peak <= 80 then 100
3512+ else max 10 (100 - (peak - 80) * 90 / 20)
3514+-- Apply adaptive scaling to a base rate limit. Reads cached counters (O(1), no bucket scans)
3515+-- and scales the rate proportionally. Returns at least 1 to prevent division-by-zero in callers.
3516+adaptiveRate : Int -> IO Int
3517+adaptiveRate = \baseRate -> do
3518+ tmc <- *totalMessageCount
3519+ sc <- *totalSessionCount
3520+ pc <- *totalPresenceCount
3521+ let scale = adaptiveRateScale tmc sc pc
3522+ yield (max 1 (baseRate * scale / 100))
3433−-- Effective rate limit readers: return runtime override if set (> 0), else compile-time default
3524+-- Effective rate limit readers: return runtime override if set (> 0), else compile-time default.
3525+-- All per-key and global rates are scaled by adaptiveRate under high utilization.
3434effectiveGlobalRequestRate = do
3526effectiveGlobalRequestRate = do
3435 o <- *rateOverrideGlobalRequest
3527 o <- *rateOverrideGlobalRequest
3436− yield (if o > 0 then o else maxGlobalRequestRate)
3528+ adaptiveRate (if o > 0 then o else maxGlobalRequestRate)
3438effectiveIpRate = do
3530effectiveIpRate = do
3439 o <- *rateOverrideIpRate
3531 o <- *rateOverrideIpRate
3440− yield (if o > 0 then o else maxIpRatePerIp)
3532+ adaptiveRate (if o > 0 then o else maxIpRatePerIp)
3442effectiveSendRate = do
3534effectiveSendRate = do
3443 o <- *rateOverrideSendRate
3535 o <- *rateOverrideSendRate
3444− yield (if o > 0 then o else maxSendRatePerKey)
3536+ adaptiveRate (if o > 0 then o else maxSendRatePerKey)
3446effectiveFederationRate = do
3538effectiveFederationRate = do
3447 o <- *rateOverrideFederationRate
3539 o <- *rateOverrideFederationRate
3540+ -- Federation rates are NOT adaptively scaled: throttling federation under load
3541+ -- prevents ack delivery, causing origin servers to retry (amplification loop).
3448 yield (if o > 0 then o else maxFederationRatePerServer)
3542 yield (if o > 0 then o else maxFederationRatePerServer)
3450effectiveAuthGlobalRate = do
3544effectiveAuthGlobalRate = do
3451 o <- *rateOverrideAuthGlobal
3545 o <- *rateOverrideAuthGlobal
3452− yield (if o > 0 then o else maxAuthGlobalRate)
3546+ adaptiveRate (if o > 0 then o else maxAuthGlobalRate)
3454effectiveRecvRate = do
3548effectiveRecvRate = do
3455 o <- *rateOverrideRecvRate
3549 o <- *rateOverrideRecvRate
3456− yield (if o > 0 then o else maxRecvRatePerKey)
3550+ adaptiveRate (if o > 0 then o else maxRecvRatePerKey)
3458effectiveChallengeRate = do
3552effectiveChallengeRate = do
3459 o <- *rateOverrideChallengeRate
3553 o <- *rateOverrideChallengeRate
3460− yield (if o > 0 then o else maxChallengeRatePerKey)
3554+ adaptiveRate (if o > 0 then o else maxChallengeRatePerKey)
3462effectivePairRate = do
3556effectivePairRate = do
3463 o <- *rateOverridePairRate
3557 o <- *rateOverridePairRate
3464− yield (if o > 0 then o else maxPairRatePerPair)
3558+ adaptiveRate (if o > 0 then o else maxPairRatePerPair)
3466effectiveLookupRate = do
3560effectiveLookupRate = do
3467 o <- *rateOverrideLookupRate
3561 o <- *rateOverrideLookupRate
3468− yield (if o > 0 then o else maxLookupRatePerKey)
3562+ adaptiveRate (if o > 0 then o else maxLookupRatePerKey)
3470effectiveClientWriteRate = do
3564effectiveClientWriteRate = do
3471 o <- *rateOverrideClientWriteRate
3565 o <- *rateOverrideClientWriteRate
3472− yield (if o > 0 then o else maxClientWriteRatePerKey)
3566+ adaptiveRate (if o > 0 then o else maxClientWriteRatePerKey)
3474checkFederationRate = \server t -> atomic do
3568checkFederationRate = \server t -> atomic do
3475 federationRates <- *federationRates
3569 federationRates <- *federationRates
@@ -4375,11 +4469,12 @@ ackPolledMessages = \pubkey cursor t -> do
4376pollNewEvents = \pubkey clampedCursor t -> do
4470pollNewEvents = \pubkey clampedCursor t -> do
4377 bucket <- readMessageBucket (pubkeyBucket pubkey)
4471 bucket <- readMessageBucket (pubkeyBucket pubkey)
4378− -- NOTE: sortBy is technically redundant since receivedAt is monotonically increasing
4472+ -- Priority-aware delivery: high-priority messages surface first within the maxPollEvents cap.
4379− -- (via nextSeqTs) and union appends new messages at the end. Kept as a defensive
4473+ -- Sort key: (-priority * a large constant + receivedAt) ensures priority dominates while
4380− -- measure in case bucket ordering assumptions are ever violated (e.g., by a future
4474+ -- preserving receivedAt order within the same priority level. The constant must exceed the
4381− -- bulk-import feature or runtime changes to list representation).
4475+ -- maximum possible receivedAt spread within one poll batch (bounded by messageMaxAge).
4476+ -- Priority 0 (default) sorts after all higher priorities.
4382− let allPollable = sortBy (\m -> m.receivedAt) (pollableMessages bucket pubkey clampedCursor t)
4477+ let allPollable = sortBy (\m -> (0 - m.priority) * 10000000000000 + m.receivedAt) (pollableMessages bucket pubkey clampedCursor t)
4383 let limitedMsgs = takeRelation maxPollEvents allPollable
4478 let limitedMsgs = takeRelation maxPollEvents allPollable
4384 let evts = do
4479 let evts = do
4385 m <- limitedMsgs
4480 m <- limitedMsgs
@@ -5895,6 +5990,7 @@ handleDrain = \phase authorization respond -> do
5895 authorized <- verifyMaintenanceAuth authorization
5990 authorized <- verifyMaintenanceAuth authorization
5896 if not authorized then yield (respond {ok: False {}, drained: 0, phase: 0})
5991 if not authorized then yield (respond {ok: False {}, drained: 0, phase: 0})
5993+ logEvent "audit" "drain" ("phase=" ++ show phase)
5898 -- Effective phase: 0 or >=2 means full drain (backward compat for clients sending empty body)
5994 -- Effective phase: 0 or >=2 means full drain (backward compat for clients sending empty body)
5899 let effectivePhase = if phase == 1 then 1 else 2
5995 let effectivePhase = if phase == 1 then 1 else 2
5900 -- Idempotency: if already at or above requested phase, return success.
5996 -- Idempotency: if already at or above requested phase, return success.
@@ -5956,6 +6052,7 @@ handleDrain = \phase authorization respond -> do
5956 println "[SNAPSHOT_END]"
6052 println "[SNAPSHOT_END]"
5957 atomic do set *prunesSinceSnapshot = 0
6053 atomic do set *prunesSinceSnapshot = 0
5958 logInfo ("drain: auto-snapshot exported (" ++ show (count drainResult.localEntries) ++ " presence, " ++ show (count flushState.forwards) ++ " forwards)")
6054 logInfo ("drain: auto-snapshot exported (" ++ show (count drainResult.localEntries) ++ " presence, " ++ show (count flushState.forwards) ++ " forwards)")
6055+ logEvent "warn" "snapshot_unencrypted" "drain snapshot exported to stdout — ensure log pipeline is secured"
5959 -- Fork best-effort flush of pending federation ops, then gossip, then wait for in-flight.
6056 -- Fork best-effort flush of pending federation ops, then gossip, then wait for in-flight.
5960 -- Operator polls /stats for drainCompletedAt > 0 to know when safe to stop.
6057 -- Operator polls /stats for drainCompletedAt > 0 to know when safe to stop.
5961 atomic do set *drainCompletedAt = 0
6058 atomic do set *drainCompletedAt = 0
@@ -5999,6 +6096,7 @@ handleUndrain = \authorization respond -> do
5999 authorized <- verifyMaintenanceAuth authorization
6096 authorized <- verifyMaintenanceAuth authorization
6000 if not authorized then yield (respond {ok: False {}})
6097 if not authorized then yield (respond {ok: False {}})
6099+ logEvent "audit" "undrain" ""
6002 wasDraining <- atomic do
6100 wasDraining <- atomic do
6003 d <- *drainPhase
6101 d <- *drainPhase
6004 set *drainPhase = 0
6102 set *drainPhase = 0
@@ -6020,6 +6118,7 @@ handleShutdown = \authorization respond -> do
6020 authorized <- verifyMaintenanceAuth authorization
6118 authorized <- verifyMaintenanceAuth authorization
6021 if not authorized then yield (respond {ok: False {}, drained: 0})
6119 if not authorized then yield (respond {ok: False {}, drained: 0})
6121+ logEvent "audit" "shutdown" ""
6023 dp <- *drainPhase
6122 dp <- *drainPhase
6024 if dp >= 2 then do
6123 if dp >= 2 then do
6025 -- Already draining, just set the shutdown flag
6124 -- Already draining, just set the shutdown flag
@@ -6048,6 +6147,7 @@ handleReconcile = \authorization respond -> do
6048 authorized <- verifyMaintenanceAuth authorization
6147 authorized <- verifyMaintenanceAuth authorization
6049 if not authorized then yield (respond {ok: False {}, messageDrift: 0, presenceDrift: 0, sessionDrift: 0})
6148 if not authorized then yield (respond {ok: False {}, messageDrift: 0, presenceDrift: 0, sessionDrift: 0})
6150+ logEvent "audit" "reconcile" ""
6051 presDrift <- atomic do
6151 presDrift <- atomic do
6052 actual <- countAllPresence
6152 actual <- countAllPresence
6053 stored <- *totalPresenceCount
6153 stored <- *totalPresenceCount
@@ -6405,6 +6505,7 @@ handleRestore = \version checksumIn presenceData ackedIdsData ackOverflowData de
6405 authorized <- verifyMaintenanceAuth authorization
6505 authorized <- verifyMaintenanceAuth authorization
6406 if not authorized then yield (respond {ok: False {}, error: "unauthorized", presenceCount: 0, ackedIdsCount: 0, deliveredIdsCount: 0})
6506 if not authorized then yield (respond {ok: False {}, error: "unauthorized", presenceCount: 0, ackedIdsCount: 0, deliveredIdsCount: 0})
6508+ logEvent "audit" "restore" ("version=" ++ show version ++ " presence=" ++ show (count presenceData) ++ " forwards=" ++ show (count forwardsData))
6408 -- Minimum version: v6 (first version with checksum integrity verification).
6509 -- Minimum version: v6 (first version with checksum integrity verification).
6409 -- Pre-v6 snapshots (0-5) have no checksum and could be corrupted without detection.
6510 -- Pre-v6 snapshots (0-5) have no checksum and could be corrupted without detection.
6410 -- Versions 6-10 are accepted with backward compatibility (missing fields default to []/0).
6511 -- Versions 6-10 are accepted with backward compatibility (missing fields default to []/0).
@@ -6861,6 +6962,7 @@ handleBlockPeer = \server authorization respond -> do
6861 authorized <- verifyMaintenanceAuth authorization
6962 authorized <- verifyMaintenanceAuth authorization
6862 if not authorized then yield (respond {ok: False {}, blocked: 0})
6963 if not authorized then yield (respond {ok: False {}, blocked: 0})
6965+ logEvent "audit" "block_peer" server
6864 result <- atomic do
6966 result <- atomic do
6865 blocked <- *blockedPeers
6967 blocked <- *blockedPeers
6866 if isBlockedPeer server blocked then yield {alreadyBlocked: True {}, total: count blocked}
6968 if isBlockedPeer server blocked then yield {alreadyBlocked: True {}, total: count blocked}
@@ -6881,6 +6983,7 @@ handleUnblockPeer = \server authorization respond -> do
6881 authorized <- verifyMaintenanceAuth authorization
6983 authorized <- verifyMaintenanceAuth authorization
6882 if not authorized then yield (respond {ok: False {}, unblocked: False {}})
6984 if not authorized then yield (respond {ok: False {}, unblocked: False {}})
6986+ logEvent "audit" "unblock_peer" server
6884 removed <- atomic do
6987 removed <- atomic do
6885 blocked <- *blockedPeers
6988 blocked <- *blockedPeers
6886 let had = isBlockedPeer server blocked
6989 let had = isBlockedPeer server blocked
@@ -6904,6 +7007,7 @@ handleAllowPeer = \server authorization respond -> do
6904 authorized <- verifyMaintenanceAuth authorization
7007 authorized <- verifyMaintenanceAuth authorization
6905 if not authorized then yield (respond {ok: False {}, allowed: 0})
7008 if not authorized then yield (respond {ok: False {}, allowed: 0})
7010+ logEvent "audit" "allow_peer" server
6907 result <- atomic do
7011 result <- atomic do
6908 allowed <- *allowedPeers
7012 allowed <- *allowedPeers
6909 if any allowed (\a -> a == server) then yield {alreadyAllowed: True {}, total: count allowed}
7013 if any allowed (\a -> a == server) then yield {alreadyAllowed: True {}, total: count allowed}
@@ -6924,6 +7028,7 @@ handleUnallowPeer = \server authorization respond -> do
6924 authorized <- verifyMaintenanceAuth authorization
7028 authorized <- verifyMaintenanceAuth authorization
6925 if not authorized then yield (respond {ok: False {}, removed: False {}})
7029 if not authorized then yield (respond {ok: False {}, removed: False {}})
7031+ logEvent "audit" "unallow_peer" server
6927 removed <- atomic do
7032 removed <- atomic do
6928 allowed <- *allowedPeers
7033 allowed <- *allowedPeers
6929 let had = any allowed (\a -> a == server)
7034 let had = any allowed (\a -> a == server)
@@ -6985,6 +7090,7 @@ handleAdminRevokeSessions = \pubkey authorization respond -> do
6985 authorized <- verifyMaintenanceAuth authorization
7090 authorized <- verifyMaintenanceAuth authorization
6986 if not authorized then yield (respond {ok: False {}, revokedCount: 0})
7091 if not authorized then yield (respond {ok: False {}, revokedCount: 0})
7093+ logEvent "audit" "revoke_all_sessions" pubkey
6988 t <- now
7094 t <- now
6989 result <- atomic do
7095 result <- atomic do
6990 -- Filter each session bucket in-place (avoids readAllSessions + writeAllSessions repartition)
7096 -- Filter each session bucket in-place (avoids readAllSessions + writeAllSessions repartition)
@@ -7016,6 +7122,7 @@ handleAdminPurgeMessages = \pubkey authorization respond -> do
7016 authorized <- verifyMaintenanceAuth authorization
7122 authorized <- verifyMaintenanceAuth authorization
7017 if not authorized then yield (respond {ok: False {}, purgedCount: 0})
7123 if not authorized then yield (respond {ok: False {}, purgedCount: 0})
7125+ logEvent "audit" "purge_messages" pubkey
7019 t <- now
7126 t <- now
7020 purgeResult <- atomic do
7127 purgeResult <- atomic do
7021 let b = pubkeyBucket pubkey
7128 let b = pubkeyBucket pubkey
@@ -7063,8 +7170,10 @@ handleAdminPurgeMessages = \pubkey authorization respond -> do
7063handleAdminRevokeIp = \ip authorization respond -> do
7170handleAdminRevokeIp = \ip authorization respond -> do
7064 authorized <- verifyMaintenanceAuth authorization
7171 authorized <- verifyMaintenanceAuth authorization
7065 if not authorized then yield (respond {ok: False {}, revokedCount: 0, affectedKeys: 0})
7172 if not authorized then yield (respond {ok: False {}, revokedCount: 0, affectedKeys: 0})
7066− else if ip == "" then yield (respond {ok: False {}, revokedCount: 0, affectedKeys: 0})
7174+ logEvent "audit" "revoke_ip" ip
7175+ if ip == "" then yield (respond {ok: False {}, revokedCount: 0, affectedKeys: 0})
7068 t <- now
7177 t <- now
7069 result <- revokeSessionsByIp ip 0 t {revokedCount: 0, affectedKeys: [], cleanups: []}
7178 result <- revokeSessionsByIp ip 0 t {revokedCount: 0, affectedKeys: [], cleanups: []}
7070 -- Gossip offline for pubkeys that lost their last session
7179 -- Gossip offline for pubkeys that lost their last session
@@ -7152,6 +7261,7 @@ handleFlushNonces = \authorization respond -> do
7152 authorized <- verifyMaintenanceAuth authorization
7261 authorized <- verifyMaintenanceAuth authorization
7153 if not authorized then yield (respond {ok: False {}, flushedCount: 0})
7262 if not authorized then yield (respond {ok: False {}, flushedCount: 0})
7264+ logEvent "audit" "flush_nonces" ""
7155 flushedCount <- atomic do
7265 flushedCount <- atomic do
7156 total <- countAllNonces 0
7266 total <- countAllNonces 0
7157 writeAllNonces []
7267 writeAllNonces []
@@ -7165,20 +7275,22 @@ handleFlushNonces = \authorization respond -> do
7165handleRotateSecret = \newSecret authorization respond -> do
7275handleRotateSecret = \newSecret authorization respond -> do
7166 authorized <- verifyMaintenanceAuth authorization
7276 authorized <- verifyMaintenanceAuth authorization
7167 if not authorized then yield (respond {ok: False {}, error: "unauthorized"})
7277 if not authorized then yield (respond {ok: False {}, error: "unauthorized"})
7168− else if newSecret != "" && federationEnabled && not (tooLong newSecret 15) then do
7169− logError "rotate-secret: rejected — secret must be at least 16 characters when federation is enabled"
7170− yield (respond {ok: False {}, error: "secret_too_short"})
7171− else if newSecret != "" && not (tooLong newSecret 31) then do
7172− logWarn "rotate-secret: new secret is shorter than 32 characters — 32+ recommended for production"
7173− atomic do set *runtimeMaintenanceSecret = newSecret
7174− logInfo "rotate-secret: maintenance secret rotated (warning: short secret)"
7175− yield (respond {ok: True {}, error: ""})
7279+ logEvent "audit" "rotate_secret" (if newSecret == "" then "revert_to_default" else "new_secret")
7280+ if newSecret != "" && federationEnabled && not (tooLong newSecret 15) then do
7281+ logError "rotate-secret: rejected — secret must be at least 16 characters when federation is enabled"
7282+ yield (respond {ok: False {}, error: "secret_too_short"})
7283+ else if newSecret != "" && not (tooLong newSecret 31) then do
7284+ logWarn "rotate-secret: new secret is shorter than 32 characters — 32+ recommended for production"
7177− atomic do set *runtimeMaintenanceSecret = newSecret
7285+ atomic do set *runtimeMaintenanceSecret = newSecret
7286+ logInfo "rotate-secret: maintenance secret rotated (warning: short secret)"
7287+ yield (respond {ok: True {}, error: ""})
7289+ atomic do set *runtimeMaintenanceSecret = newSecret
7178− if newSecret == ""
7290+ if newSecret == ""
7179− then logInfo "rotate-secret: reverted to compile-time maintenance secret"
7291+ then logInfo "rotate-secret: reverted to compile-time maintenance secret"
7180− else logInfo "rotate-secret: maintenance secret rotated"
7292+ else logInfo "rotate-secret: maintenance secret rotated"
7181− yield (respond {ok: True {}, error: ""})
7293+ yield (respond {ok: True {}, error: ""})
7183-- GET /maintenance/rate-limits: Return current effective rate limits.
7295-- GET /maintenance/rate-limits: Return current effective rate limits.
7184handleGetRateLimits = \authorization respond -> do
7296handleGetRateLimits = \authorization respond -> do
@@ -7205,6 +7317,7 @@ handleSetRateLimits = \globalRequest ipRate sendRate federationRate authGlobal r
7205 authorized <- verifyMaintenanceAuth authorization
7317 authorized <- verifyMaintenanceAuth authorization
7206 if not authorized then yield (respond {ok: False {}})
7318 if not authorized then yield (respond {ok: False {}})
7320+ logEvent "audit" "set_rate_limits" ("global=" ++ show globalRequest ++ " ip=" ++ show ipRate ++ " send=" ++ show sendRate)
7208 atomic do
7321 atomic do
7209 set *rateOverrideGlobalRequest = max 0 globalRequest
7322 set *rateOverrideGlobalRequest = max 0 globalRequest
7210 set *rateOverrideIpRate = max 0 ipRate
7323 set *rateOverrideIpRate = max 0 ipRate
@@ -7390,13 +7503,86 @@ serveInner = \req -> case req of
7390 TopRecipients {limit, authorization, respond} -> handleTopRecipients limit authorization respond
7503 TopRecipients {limit, authorization, respond} -> handleTopRecipients limit authorization respond
7391 Ready {respond} -> handleReady respond
7504 Ready {respond} -> handleReady respond
7506+-- Backpressure: check if the server is overloaded (too many concurrent requests).
7507+-- Health, Ready, and Maintenance endpoints are exempt so operators can always monitor
7508+-- and control the server even under extreme load.
7509+isOverloadExempt = \req -> case req of
7510+ Health {} -> True {}
7511+ Ready {} -> True {}
7512+ Prune {} -> True {}
7513+ Drain {} -> True {}
7514+ Shutdown {} -> True {}
7515+ Undrain {} -> True {}
7516+ Snapshot {} -> True {}
7517+ Restore {} -> True {}
7518+ Reconcile {} -> True {}
7519+ Stats {} -> True {}
7520+ StatsPeers {} -> True {}
7521+ StatsKey {} -> True {}
7524+-- Reject overloaded requests with appropriate empty/error responses per route type.
7525+-- Each response matches the handler's return shape with zeroed/empty fields.
7526+serveOverloaded = \req -> case req of
7527+ ReqChallenge {respond} -> respond {challenge: "", ok: False {}, expiresAt: 0, retryAfterMs: 1000, error: "overloaded"}
7528+ VerifyAuth {respond} -> respond {token: "", expiresAt: 0, error: "overloaded"}
7529+ RevokeSession {respond} -> respond {ok: False {}, authorized: False {}}
7530+ Logout {respond} -> respond {ok: False {}, authorized: False {}}
7531+ Poll {respond} -> respond {events: [], cursor: 0, authorized: False {}, rateLimited: True {}, draining: False {}, generation: 0, presenceUpdates: [], heartbeat: False {}, retryAfterMs: 1000, cursorReset: False {}, hasMore: False {}, cursorStale: False {}, messageExpirations: [], messageDeliveries: []}
7532+ SendMessage {respond} -> respond {status: MsRejected {}, messageId: "", retryAfterMs: 1000, error: "overloaded"}
7533+ MessageStatus {respond} -> respond {status: "", authorized: False {}}
7534+ AckMessages {respond} -> respond {ok: False {}, authorized: False {}, ackedCount: 0, error: "overloaded"}
7535+ SendBatch {respond} -> respond {results: [], authorized: False {}, retryAfterMs: 1000}
7536+ ListMessages {respond} -> respond {messages: [], authorized: False {}, count: 0, error: "overloaded"}
7537+ DeleteMessages {respond} -> respond {ok: False {}, authorized: False {}, deletedCount: 0, error: "overloaded"}
7538+ CancelMessage {respond} -> respond {ok: False {}, authorized: False {}, cancelled: False {}, error: "overloaded"}
7539+ CancelBatch {respond} -> respond {results: [], authorized: False {}, error: "overloaded"}
7540+ LookupPresence {respond} -> respond {server: "", authorized: False {}, error: "overloaded"}
7541+ BatchLookupPresence {respond} -> respond {results: [], authorized: False {}, error: "overloaded"}
7542+ RecvGossip {respond} -> respond {ok: False {}, error: "overloaded"}
7543+ ForwardMessage {respond} -> respond {ok: False {}, error: "overloaded"}
7544+ FederationAck {respond} -> respond {ok: False {}, error: "overloaded", ackedCount: 0, results: []}
7545+ FederationCancel {respond} -> respond {ok: False {}, error: "overloaded", cancelled: False {}}
7546+ FederationBatchCancel {respond} -> respond {ok: False {}, error: "overloaded", results: []}
7547+ FederationBatchForward {respond} -> respond {ok: False {}, error: "overloaded", results: []}
7548+ VerifyToken {respond} -> respond {ok: False {}}
7549+ TopRecipients {respond} -> respond {recipients: [], ok: False {}, total: 0}
7550+ BlockPeer {respond} -> respond {ok: False {}, blocked: 0}
7551+ UnblockPeer {respond} -> respond {ok: False {}, unblocked: False {}}
7552+ ListBlocked {respond} -> respond {peers: [], ok: False {}}
7553+ PingPeer {respond} -> respond {ok: False {}, latencyMs: 0, error: "overloaded"}
7554+ AdminListSessions {respond} -> respond {sessions: [], count: 0, ok: False {}}
7555+ AdminRevokeSessions {respond} -> respond {ok: False {}, revokedCount: 0}
7556+ AdminPurgeMessages {respond} -> respond {ok: False {}, purgedCount: 0}
7557+ AdminRevokeIp {respond} -> respond {ok: False {}, revokedCount: 0, affectedKeys: 0}
7558+ AdminListAllSessions {respond} -> respond {sessions: [], ok: False {}, total: 0}
7559+ GetRateLimits {respond} -> respond {ok: False {}, globalRequest: 0, ipRate: 0, sendRate: 0, federationRate: 0, authGlobal: 0, recvRate: 0, challengeRate: 0, pairRate: 0, lookupRate: 0, clientWriteRate: 0}
7560+ SetRateLimits {respond} -> respond {ok: False {}}
7561+ FlushNonces {respond} -> respond {ok: False {}, flushedCount: 0}
7562+ AllowPeer {respond} -> respond {ok: False {}, allowed: 0}
7563+ UnallowPeer {respond} -> respond {ok: False {}, removed: False {}}
7564+ ListAllowed {respond} -> respond {peers: [], ok: False {}}
7565+ RotateSecret {respond} -> respond {ok: False {}, error: "overloaded"}
7566+ -- Exempt routes handled by isOverloadExempt; this branch is unreachable
7393-- Access-logged request handler: wraps serveInner with timing and correlation ID.
7569-- Access-logged request handler: wraps serveInner with timing and correlation ID.
7394-- Logs all non-poll requests with a 6-hex-char request ID for log correlation.
7570-- Logs all non-poll requests with a 6-hex-char request ID for log correlation.
7395-- Polls are only logged when slow (> slowRequestThreshold) to avoid overwhelming
7571-- Polls are only logged when slow (> slowRequestThreshold) to avoid overwhelming
7396-- logs with high-frequency long-poll traffic.
7572-- logs with high-frequency long-poll traffic.
7573+-- Backpressure: rejects non-exempt requests when activeRequests >= maxActiveRequests.
7397serve = \req -> do
7574serve = \req -> do
7398 startTime <- now
7575 startTime <- now
7399 reqId <- genTokenN 1
7576 reqId <- genTokenN 1
7577+ -- Backpressure check: reject early if overloaded (exempt: health, ready, maintenance)
7578+ overloaded <- if maxActiveRequests <= 0 || isOverloadExempt req then yield (False {})
7580+ ar <- *activeRequests
7581+ yield (ar >= maxActiveRequests)
7582+ if overloaded then do
7583+ logEvent "warn" "overloaded" (requestPath req)
7584+ serveOverloaded req
7400 atomic do
7586 atomic do
7401 ar <- *activeRequests
7587 ar <- *activeRequests
7402 let newAr = ar + 1
7588 let newAr = ar + 1
@@ -7659,6 +7845,12 @@ backgroundWatchdog = do
7659 atomic do
7845 atomic do
7660 inFlight <- *federationInFlight
7846 inFlight <- *federationInFlight
7661 set *federationActiveConnections = fold (\acc r -> acc + r.count) 0 inFlight)
7847 set *federationActiveConnections = fold (\acc r -> acc + r.count) 0 inFlight)
7848+ -- Log hung federation connections even below the max threshold (early warning)
7849+ inFlight <- *federationInFlight
7850+ let hungSlots = do r <- inFlight; where r.count > 0 && r.firstStartedAt > 0 && t - r.firstStartedAt >= federationFetchTimeoutMs; yield r
7851+ when (count hungSlots > 0) (do
7852+ logEvent "warn" "watchdog_hung_federation" (show (count hungSlots) ++ " servers with hung connections")
7853+ logStaleInFlight hungSlots t)
7662 backgroundWatchdog
7854 backgroundWatchdog
7664-- Startup connectivity probe: send an empty gossip to each known peer and log results.
7856-- Startup connectivity probe: send an empty gossip to each known peer and log results.
@@ -7753,6 +7945,7 @@ main = do
7753 when (federationEnabled && isBadServerName serverHostN)
7945 when (federationEnabled && isBadServerName serverHostN)
7754 (logError ("serverHost '" ++ serverHost ++ "' would be rejected by isBadServerName — federation peers cannot forward messages here. Set serverHost to a public DNS name."))
7946 (logError ("serverHost '" ++ serverHost ++ "' would be rejected by isBadServerName — federation peers cannot forward messages here. Set serverHost to a public DNS name."))
7755 logInfo "IP binding is active — sessions created with a ClientIp header will reject requests without one"
7947 logInfo "IP binding is active — sessions created with a ClientIp header will reject requests without one"
7948+ logWarn "SECURITY: Knot has no HTTP-layer request body size limit. Deploy behind a reverse proxy with a body limit (e.g., Caddy: request_body { max_size 256KB }) to prevent large-payload DoS."
7756 when federationEnabled
7949 when federationEnabled
7757 (logWarn "SECURITY: DNS rebinding risk — federation URLs resolve DNS at fetch time, not at validation time. A peer hostname passing isBadServerName could rebind to a private IP between validation and fetch. Mitigate by blocking outbound connections to RFC1918/link-local/CGNAT ranges at the firewall level.")
7950 (logWarn "SECURITY: DNS rebinding risk — federation URLs resolve DNS at fetch time, not at validation time. A peer hostname passing isBadServerName could rebind to a private IP between validation and fetch. Mitigate by blocking outbound connections to RFC1918/link-local/CGNAT ranges at the firewall level.")
7758 logInfo ("Skrepka relay server starting on port " ++ show listenPort ++ "...")
7951 logInfo ("Skrepka relay server starting on port " ++ show listenPort ++ "...")
@@ -7765,6 +7958,7 @@ main = do
7765 logInfo (" max forwards: " ++ show maxForwardsEntries ++ " (per key: " ++ show maxForwardsPerKey ++ ")")
7958 logInfo (" max forwards: " ++ show maxForwardsEntries ++ " (per key: " ++ show maxForwardsPerKey ++ ")")
7766 logInfo (" session expiry: " ++ show sessionExpiry ++ "ms (max lifetime: " ++ show maxSessionLifetime ++ "ms)")
7959 logInfo (" session expiry: " ++ show sessionExpiry ++ "ms (max lifetime: " ++ show maxSessionLifetime ++ "ms)")
7767 logInfo (" message max age: " ++ show messageMaxAge ++ "ms")
7960 logInfo (" message max age: " ++ show messageMaxAge ++ "ms")
7961+ logInfo (" max active requests: " ++ if maxActiveRequests > 0 then show maxActiveRequests else "unlimited")
7768 logInfo (" global rate limit: " ++ show maxGlobalRequestRate ++ "/" ++ show globalRateWindowMs ++ "ms")
7962 logInfo (" global rate limit: " ++ show maxGlobalRequestRate ++ "/" ++ show globalRateWindowMs ++ "ms")
7769 logInfo (" auth rate limit: " ++ show maxAuthGlobalRate ++ "/" ++ show authGlobalRateWindowMs ++ "ms")
7963 logInfo (" auth rate limit: " ++ show maxAuthGlobalRate ++ "/" ++ show authGlobalRateWindowMs ++ "ms")
7770 logInfo (" pair rate limit: " ++ show maxPairRatePerPair ++ "/" ++ show pairRateWindowMs ++ "ms per sender-recipient")
7964 logInfo (" pair rate limit: " ++ show maxPairRatePerPair ++ "/" ++ show pairRateWindowMs ++ "ms per sender-recipient")