7147117 · skrepka · git.iko.soy

git.iko.soy/skrepka

on master help sign in

Simplify server: unify gateAuth, extract authPreCheck, inline forwardForNewlyOnline, refactor helpers

7147117iko6 months ago

server.knot+62−68

compare with master

unifiedsplit

@@ -513,14 +513,14 @@ forEach = \items getKey action -> case headOf items of

513 Nothing {} -> yield {}

514 Just {value: x} -> do

515 action x

516− forEach (do y <- items; where getKey y != getKey x; yield y) getKey action

516+ forEach (filter items (\y -> getKey y != getKey x)) getKey action

517

518-- Like forEach, but collects each action's result into a list.

519mapEach = \items getKey action -> case headOf items of

520 Nothing {} -> yield []

521 Just {value: x} -> do

522 r <- action x

523− rest <- mapEach (do y <- items; where getKey y != getKey x; yield y) getKey action

523+ rest <- mapEach (filter items (\y -> getKey y != getKey x)) getKey action

524 yield (union [r] rest)

525

526-- Hex tokens

@@ -612,22 +612,16 @@ withMaintAuth = \authorization onUnauth onAuth -> do

612 authorized <- verifyMaintenanceAuth authorization

613 if authorized then onAuth else onUnauth

614

615−-- Common gate for client handlers: drain check, global rate, session auth.

615+-- Common gate for handlers: optional drain check, global rate, session auth.

616-- mkErr is called with one of: "draining", "rate_limited", "unauthorized".

617−gateAuth = \auth ip t respond mkErr onAuth -> do

617+gateAuth = \checkDrain auth ip t respond mkErr onAuth -> do

618 dp <- *drainPhase

619− if dp >= 2 then yield (respond (mkErr "draining"))

619+ if checkDrain && dp >= 2 then yield (respond (mkErr "draining"))

620 else do

621 g <- checkGlobalRate t

622 if g then yield (respond (mkErr "rate_limited"))

623 else withSessionAuth auth t ip (yield (respond (mkErr "unauthorized"))) onAuth

624

625−-- Variant without the drain check (used by lookup/status handlers).

626−gateAuthNoDrain = \auth ip t respond mkErr onAuth -> do

627− g <- checkGlobalRate t

628− if g then yield (respond (mkErr "rate_limited"))

629− else withSessionAuth auth t ip (yield (respond (mkErr "unauthorized"))) onAuth

630−

631authenticateSession : [Session] -> Text -> Timestamp -> AuthResult

625authenticateSession : [Session] -> Text -> Timestamp -> AuthResult

632authenticateSession = \sessions token t ->

626authenticateSession = \sessions token t ->

633 case findFirst sessions (\s -> s.expiresAt > t && s.token == token) of

627 case findFirst sessions (\s -> s.expiresAt > t && s.token == token) of

@@ -764,6 +758,14 @@ checkPreAuth = \clientIp t -> do

764 else if i then "ip_rate_limited"

758 else if i then "ip_rate_limited"

765 else "")

759 else "")

766

760

761+-- Pre-auth check + auth-global rate. Returns "" when ok, else an error code.

762+authPreCheck = \clientIp t -> do

763+ p <- checkPreAuth clientIp t

764+ if p != "" then yield p

765+ else do

766+ a <- checkAuthGlobalRate t

767+ yield (if a then "auth_rate_limited" else "")

768+

767-- ============================================

769-- ============================================

768-- Federation failure tracking (simple, no circuit breaker)

770-- Federation failure tracking (simple, no circuit breaker)

769-- ============================================

771-- ============================================

@@ -856,8 +858,9 @@ pruneExpired = \t -> atomic do

856 -- presence: remember locally-expired keys for offline gossip

858 -- presence: remember locally-expired keys for offline gossip

857 pres <- *presence

859 pres <- *presence

858 let keptPres = filter pres (\p -> p.expiresAt > t)

860 let keptPres = filter pres (\p -> p.expiresAt > t)

861+ let stillLocal = filter keptPres (\k -> isLocalServer k.server)

859 set *presence = keptPres

862 set *presence = keptPres

860− yield (filter pres (\p -> p.expiresAt <= t && isLocalServer p.server && none keptPres (\k -> k.pubkey == p.pubkey && isLocalServer k.server)))

863+ yield (filter pres (\p -> p.expiresAt <= t && isLocalServer p.server && none stillLocal (\k -> k.pubkey == p.pubkey)))

861

864

862-- ============================================

865-- ============================================

863-- Federation gossip

866-- Federation gossip

@@ -996,14 +999,11 @@ refreshLocalPresence = \t -> do

996

999

997handleChallenge = \pubkey clientIp respond -> do

1000handleChallenge = \pubkey clientIp respond -> do

998 t <- now

1001 t <- now

1002+ let mkResp = \err -> respond {challenge: "", ok: False {}, expiresAt: 0, retryAfterMs: 0, error: err}

999− if requireClientIp && clientIp == ""

1003+ if requireClientIp && clientIp == "" then yield (mkResp "client_ip_required")

1000− then yield (respond {challenge: "", ok: False {}, expiresAt: 0, retryAfterMs: 0, error: "client_ip_required"})

1001− else do

1002− preAuth <- checkPreAuth clientIp t

1003− if preAuth != "" then yield (respond {challenge: "", ok: False {}, expiresAt: 0, retryAfterMs: 0, error: preAuth})

1004 else do

1005− authLimited <- checkAuthGlobalRate t

1005+ preErr <- authPreCheck clientIp t

1006− if authLimited then yield (respond {challenge: "", ok: False {}, expiresAt: 0, retryAfterMs: 0, error: "auth_rate_limited"})

1006+ if preErr != "" then yield (mkResp preErr)

1007 else do

1008 challenge <- genToken

1009 result <- atomic do

@@ -1025,44 +1025,41 @@ handleChallenge = \pubkey clientIp respond -> do

1025

1026handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp respond -> do

1027 t <- now

1028− preAuth <- checkPreAuth clientIp t

1028+ preErr <- authPreCheck clientIp t

1029− if preAuth != "" then yield (respond {token: "", expiresAt: 0, error: preAuth})

1029+ if preErr != "" then yield (respond {token: "", expiresAt: 0, error: preErr})

1030 else do

1031− authLimited <- checkAuthGlobalRate t

1032− if authLimited then yield (respond {token: "", expiresAt: 0, error: "auth_rate_limited"})

1033− else do

1034− let expiry = t + sessionExpiry

1031+ let expiry = t + sessionExpiry

1035− token <- genToken

1032+ token <- genToken

1036− result <- atomic do

1033+ result <- atomic do

1037− ch <- *challenges

1034+ ch <- *challenges

1038− let hasMatch = any ch (\c -> c.pubkey == pubkey && c.challenge == challenge && c.expiresAt > t && matchIp clientIp c.ip)

1035+ let hasMatch = any ch (\c -> c.pubkey == pubkey && c.challenge == challenge && c.expiresAt > t && matchIp clientIp c.ip)

1039− if not hasMatch then yield {valid: False {}, peers: []}

1036+ if not hasMatch then yield {valid: False {}, peers: []}

1040− else

1037+ else

1041− if not (verify (bytesFromHex pubkey) (textToBytes challenge) (bytesFromHex signature))

1038+ if not (verify (bytesFromHex pubkey) (textToBytes challenge) (bytesFromHex signature))

1039+ then yield {valid: False {}, peers: []}

1040+ else do

1041+ set *challenges = removeWhere ch (\c -> c.pubkey == pubkey && c.challenge == challenge)

1042+ dp <- *drainPhase

1042− then yield {valid: False {}, peers: []}

1043+ if dp >= 1 then yield {valid: False {}, peers: []}

1043 else do

1044 else do

1044− set *challenges = removeWhere ch (\c -> c.pubkey == pubkey && c.challenge == challenge)

1045− dp <- *drainPhase

1046− if dp >= 1 then yield {valid: False {}, peers: []}

1047− else do

1048− ses <- *sessions

1045+ ses <- *sessions

1049− let kept = if revokeOthers then removeWhere ses (\s -> s.pubkey == pubkey) else ses

1046+ let kept = if revokeOthers then removeWhere ses (\s -> s.pubkey == pubkey) else ses

1050− let forKey = filter kept (\s -> s.pubkey == pubkey)

1047+ let forKey = filter kept (\s -> s.pubkey == pubkey)

1051− let afterPerKey = if count forKey >= maxSessionsPerKey

1048+ let afterPerKey = if count forKey >= maxSessionsPerKey

1052− then union (removeWhere kept (\s -> s.pubkey == pubkey)) (evictOldest forKey (\s -> s.expiresAt))

1049+ then union (removeWhere kept (\s -> s.pubkey == pubkey)) (evictOldest forKey (\s -> s.expiresAt))

1053− else kept

1050+ else kept

1054− let newSession = {token: token, pubkey: pubkey, expiresAt: expiry, ip: clientIp, createdAt: t}

1051+ let newSession = {token: token, pubkey: pubkey, expiresAt: expiry, ip: clientIp, createdAt: t}

1055− set *sessions = capList (union afterPerKey [newSession]) maxSessions (\s -> s.expiresAt)

1052+ set *sessions = capList (union afterPerKey [newSession]) maxSessions (\s -> s.expiresAt)

1056− pres <- *presence

1053+ pres <- *presence

1057− let withoutLocal = removeWhere pres (\p -> p.pubkey == pubkey && isLocalServer p.server)

1054+ let withoutLocal = removeWhere pres (\p -> p.pubkey == pubkey && isLocalServer p.server)

1058− let newPres = capList (union withoutLocal [{pubkey: pubkey, server: serverHostN, expiresAt: expiry, timestamp: t}]) maxPresenceEntries (\p -> p.expiresAt)

1055+ let newPres = capList (union withoutLocal [{pubkey: pubkey, server: serverHostN, expiresAt: expiry, timestamp: t}]) maxPresenceEntries (\p -> p.expiresAt)

1059− set *presence = newPres

1056+ set *presence = newPres

1060− yield {valid: True {}, peers: knownPeerServersFromPresence newPres}

1057+ yield {valid: True {}, peers: knownPeerServersFromPresence newPres}

1061− if not result.valid then yield (respond {token: "", expiresAt: 0, error: "invalid"})

1058+ if not result.valid then yield (respond {token: "", expiresAt: 0, error: "invalid"})

1062− else do

1059+ else do

1063− when federationEnabled

1060+ when federationEnabled

1064− (fork (sendBatchedGossip result.peers [mkGossip (GossipOnline {}) pubkey t onlineGossipTtl] gossipOriginHops))

1061+ (fork (sendBatchedGossip result.peers [mkGossip (GossipOnline {}) pubkey t onlineGossipTtl] gossipOriginHops))

1065− yield (respond {token: token, expiresAt: expiry, error: ""})

1062+ yield (respond {token: token, expiresAt: expiry, error: ""})

1066

1063

1067handleRevoke = \targetToken authorization clientIp respond -> do

1064handleRevoke = \targetToken authorization clientIp respond -> do

1068 t <- now

1065 t <- now

@@ -1306,7 +1303,7 @@ forkForwardsTo = \effMsgId to encryptedBlob clampedTs servers ->

1306handleLookupPresence = \pubkey authorization clientIp respond -> do

1303handleLookupPresence = \pubkey authorization clientIp respond -> do

1307 t <- now

1304 t <- now

1308 let mkErr = \e -> {server: "", authorized: False {}, error: if e == "rate_limited" then "rate_limited" else ""}

1305 let mkErr = \e -> {server: "", authorized: False {}, error: if e == "rate_limited" then "rate_limited" else ""}

1309− gateAuthNoDrain authorization clientIp t respond mkErr (\senderKey -> do

1306+ gateAuth (False {}) authorization clientIp t respond mkErr (\senderKey -> do

1310 lim <- checkPubkeyRateGate lookupRates senderKey maxLookupRatePerKey lookupRateWindowMs t

1307 lim <- checkPubkeyRateGate lookupRates senderKey maxLookupRatePerKey lookupRateWindowMs t

1311 if lim then yield (respond {server: "", authorized: True {}, error: "rate_limited"})

1308 if lim then yield (respond {server: "", authorized: True {}, error: "rate_limited"})

1312 else do

1309 else do

@@ -1318,7 +1315,7 @@ handleLookupPresence = \pubkey authorization clientIp respond -> do

1318handleBatchLookupPresence = \pubkeys authorization clientIp respond -> do

1315handleBatchLookupPresence = \pubkeys authorization clientIp respond -> do

1319 t <- now

1316 t <- now

1320 let mkErr = \e -> {results: [], authorized: False {}, error: if e == "rate_limited" then "rate_limited" else ""}

1317 let mkErr = \e -> {results: [], authorized: False {}, error: if e == "rate_limited" then "rate_limited" else ""}

1321− gateAuthNoDrain authorization clientIp t respond mkErr (\senderKey -> do

1318+ gateAuth (False {}) authorization clientIp t respond mkErr (\senderKey -> do

1322 lim <- checkPubkeyRateGate lookupRates senderKey maxLookupRatePerKey lookupRateWindowMs t

1319 lim <- checkPubkeyRateGate lookupRates senderKey maxLookupRatePerKey lookupRateWindowMs t

1323 if lim then yield (respond {results: [], authorized: True {}, error: "rate_limited"})

1320 if lim then yield (respond {results: [], authorized: True {}, error: "rate_limited"})

1324 else do

1321 else do

@@ -1333,7 +1330,7 @@ handleBatchLookupPresence = \pubkeys authorization clientIp respond -> do

1333handleMessageStatus = \to messageId authorization clientIp respond -> do

1330handleMessageStatus = \to messageId authorization clientIp respond -> do

1334 t <- now

1331 t <- now

1335 let mkErr = \_ -> {status: "", authorized: False {}}

1332 let mkErr = \_ -> {status: "", authorized: False {}}

1336− gateAuthNoDrain authorization clientIp t respond mkErr (\_ -> do

1333+ gateAuth (False {}) authorization clientIp t respond mkErr (\_ -> do

1337 msgs <- *messages

1334 msgs <- *messages

1338 fwds <- *forwards

1335 fwds <- *forwards

1339 aids <- *ackedIds

1336 aids <- *ackedIds

@@ -1352,7 +1349,7 @@ handleMessageStatus = \to messageId authorization clientIp respond -> do

1352handleAckMessages = \messageIds authorization clientIp respond -> do

1349handleAckMessages = \messageIds authorization clientIp respond -> do

1353 t <- now

1350 t <- now

1354 let mkErr = \e -> {ok: False {}, authorized: False {}, ackedCount: 0, error: e}

1351 let mkErr = \e -> {ok: False {}, authorized: False {}, ackedCount: 0, error: e}

1355− gateAuth authorization clientIp t respond mkErr (\pubkey -> do

1352+ gateAuth (True {}) authorization clientIp t respond mkErr (\pubkey -> do

1356 lim <- checkPubkeyRateGate clientWriteRates pubkey maxClientWriteRatePerKey clientWriteRateWindowMs t

1353 lim <- checkPubkeyRateGate clientWriteRates pubkey maxClientWriteRatePerKey clientWriteRateWindowMs t

1357 if lim then yield (respond {ok: False {}, authorized: True {}, ackedCount: 0, error: "rate_limited"})

1354 if lim then yield (respond {ok: False {}, authorized: True {}, ackedCount: 0, error: "rate_limited"})

1358 else do

1355 else do

@@ -1371,7 +1368,7 @@ handleAckMessages = \messageIds authorization clientIp respond -> do

1371handleListMessages = \offset authorization clientIp respond -> do

1368handleListMessages = \offset authorization clientIp respond -> do

1372 t <- now

1369 t <- now

1373 let mkErr = \e -> {messages: [], authorized: False {}, count: 0, error: if e == "rate_limited" then "rate_limited" else ""}

1370 let mkErr = \e -> {messages: [], authorized: False {}, count: 0, error: if e == "rate_limited" then "rate_limited" else ""}

1374− gateAuth authorization clientIp t respond mkErr (\pubkey -> do

1371+ gateAuth (True {}) authorization clientIp t respond mkErr (\pubkey -> do

1375 lim <- checkPubkeyRateGate lookupRates pubkey maxLookupRatePerKey lookupRateWindowMs t

1372 lim <- checkPubkeyRateGate lookupRates pubkey maxLookupRatePerKey lookupRateWindowMs t

1376 if lim then yield (respond {messages: [], authorized: True {}, count: 0, error: "rate_limited"})

1373 if lim then yield (respond {messages: [], authorized: True {}, count: 0, error: "rate_limited"})

1377 else do

1374 else do

@@ -1386,7 +1383,7 @@ handleListMessages = \offset authorization clientIp respond -> do

1386handleDeleteMessages = \messageIds authorization clientIp respond -> do

1383handleDeleteMessages = \messageIds authorization clientIp respond -> do

1387 t <- now

1384 t <- now

1388 let mkErr = \e -> {ok: False {}, authorized: False {}, deletedCount: 0, error: if e == "draining" then "" else e}

1385 let mkErr = \e -> {ok: False {}, authorized: False {}, deletedCount: 0, error: if e == "draining" then "" else e}

1389− gateAuth authorization clientIp t respond mkErr (\pubkey -> do

1386+ gateAuth (True {}) authorization clientIp t respond mkErr (\pubkey -> do

1390 lim <- checkPubkeyRateGate clientWriteRates pubkey maxClientWriteRatePerKey clientWriteRateWindowMs t

1387 lim <- checkPubkeyRateGate clientWriteRates pubkey maxClientWriteRatePerKey clientWriteRateWindowMs t

1391 if lim then yield (respond {ok: False {}, authorized: True {}, deletedCount: 0, error: "rate_limited"})

1388 if lim then yield (respond {ok: False {}, authorized: True {}, deletedCount: 0, error: "rate_limited"})

1392 else do

1389 else do

@@ -1406,7 +1403,7 @@ handleDeleteMessages = \messageIds authorization clientIp respond -> do

1406handleCancelMessage = \messageId to authorization clientIp respond -> do

1403handleCancelMessage = \messageId to authorization clientIp respond -> do

1407 t <- now

1404 t <- now

1408 let mkErr = \e -> {ok: False {}, authorized: False {}, cancelled: False {}, error: if e == "draining" then "" else e}

1405 let mkErr = \e -> {ok: False {}, authorized: False {}, cancelled: False {}, error: if e == "draining" then "" else e}

1409− gateAuth authorization clientIp t respond mkErr (\senderKey -> do

1406+ gateAuth (True {}) authorization clientIp t respond mkErr (\senderKey -> do

1410 lim <- checkPubkeyRateGate clientWriteRates senderKey maxClientWriteRatePerKey clientWriteRateWindowMs t

1407 lim <- checkPubkeyRateGate clientWriteRates senderKey maxClientWriteRatePerKey clientWriteRateWindowMs t

1411 if lim then yield (respond {ok: False {}, authorized: True {}, cancelled: False {}, error: "rate_limited"})

1408 if lim then yield (respond {ok: False {}, authorized: True {}, cancelled: False {}, error: "rate_limited"})

1412 else do

1409 else do

@@ -1446,7 +1443,7 @@ sendCancelToEach = \messageId to servers ->

1446handleCancelBatch = \cancels authorization clientIp respond -> do

1443handleCancelBatch = \cancels authorization clientIp respond -> do

1447 t <- now

1444 t <- now

1448 let mkErr = \e -> {results: [], authorized: False {}, error: if e == "draining" then "" else e}

1445 let mkErr = \e -> {results: [], authorized: False {}, error: if e == "draining" then "" else e}

1449− gateAuth authorization clientIp t respond mkErr (\senderKey -> do

1446+ gateAuth (True {}) authorization clientIp t respond mkErr (\senderKey -> do

1450 lim <- checkPubkeyRateGate clientWriteRates senderKey maxClientWriteRatePerKey clientWriteRateWindowMs t

1447 lim <- checkPubkeyRateGate clientWriteRates senderKey maxClientWriteRatePerKey clientWriteRateWindowMs t

1451 if lim then yield (respond {results: [], authorized: True {}, error: "rate_limited"})

1448 if lim then yield (respond {results: [], authorized: True {}, error: "rate_limited"})

1452 else do

1449 else do

@@ -1470,7 +1467,7 @@ handleCancelBatch = \cancels authorization clientIp respond -> do

1470handleSendBatch = \messages authorization clientIp respond -> do

1467handleSendBatch = \messages authorization clientIp respond -> do

1471 t <- now

1468 t <- now

1472 let mkErr = \_ -> {results: [], authorized: False {}, retryAfterMs: 0}

1469 let mkErr = \_ -> {results: [], authorized: False {}, retryAfterMs: 0}

1473− gateAuth authorization clientIp t respond mkErr (\senderKey -> do

1470+ gateAuth (True {}) authorization clientIp t respond mkErr (\senderKey -> do

1474 results <- mapEach (takeRelation maxBatchSize messages) id (\m -> do

1471 results <- mapEach (takeRelation maxBatchSize messages) id (\m -> do

1475 effMsgId <- if m.messageId != "" then yield m.messageId else genToken

1472 effMsgId <- if m.messageId != "" then yield m.messageId else genToken

1476 let clampedTs = clampTimestamp m.timestamp t

1473 let clampedTs = clampTimestamp m.timestamp t

@@ -1528,12 +1525,9 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp correlation

1528 let newlyOnline = mapWhere online (\e -> not (isLocalServer e.server) && none pres (\p -> p.pubkey == e.pubkey && p.server == e.server && p.expiresAt >= e.expiresAt)) (\e -> {pubkey: e.pubkey, server: e.server})

1525 let newlyOnline = mapWhere online (\e -> not (isLocalServer e.server) && none pres (\p -> p.pubkey == e.pubkey && p.server == e.server && p.expiresAt >= e.expiresAt)) (\e -> {pubkey: e.pubkey, server: e.server})

1529 yield {newlyOnline: newlyOnline}

1526 yield {newlyOnline: newlyOnline}

1530 when (count valid.newlyOnline > 0)

1527 when (count valid.newlyOnline > 0)

1531− (fork (forwardForNewlyOnline valid.newlyOnline t))

1528+ (fork (forEach valid.newlyOnline (\tgt -> tgt.pubkey ++ "|" ++ tgt.server) (\tgt -> forwardQueuedToServer tgt.pubkey tgt.server t)))

1532 yield (respond {ok: True {}, error: "", retryAfterMs: 0}))

1529 yield (respond {ok: True {}, error: "", retryAfterMs: 0}))

1533

1530

1534−forwardForNewlyOnline = \targets t ->

1535− forEach targets (\tgt -> tgt.pubkey ++ "|" ++ tgt.server) (\tgt -> forwardQueuedToServer tgt.pubkey tgt.server t)

1536−

1537handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fromServer requestTimestamp correlationId expiresAt priority authorization respond -> do

1531handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fromServer requestTimestamp correlationId expiresAt priority authorization respond -> do

1538 t <- now

1532 t <- now

1539 withFederationAuth authorization t respond fromServer ("forward:" ++ messageId) requestTimestamp (not (isLocalServer originServer) && not (isLocalServer fromServer)) (do

1533 withFederationAuth authorization t respond fromServer ("forward:" ++ messageId) requestTimestamp (not (isLocalServer originServer) && not (isLocalServer fromServer)) (do