3b2ae7b · skrepka · git.iko.soy

git.iko.soy/skrepka

on master help sign in

Stop deleting messages on FederationAck to prevent message loss

3b2ae7biko7 months ago

server.knot+63−36

compare with master

unifiedsplit

@@ -27,6 +27,11 @@ maxMessageIdLen = 128

27maxTokenLen = 256

28maxChallenges = 10000

29maxSessions = 10000

30+maxChallengesPerKey = 5

31+maxSessionsPerKey = 10

32+challengeExpiry = 60000 -- 1 minute in milliseconds

33+sessionExpiry = 3600000 -- 1 hour in milliseconds

34+onlineGossipTtl = 3600 -- 1 hour in seconds

31-- ============================================

36-- ============================================

32-- Types

37-- Types

@@ -241,17 +246,7 @@ stripTrailingDot = \s ->

241 else if take 1 s == "." && isTrailingDots (drop 1 s) then stripTrailingDot (drop 1 s)

246 else if take 1 s == "." && isTrailingDots (drop 1 s) then stripTrailingDot (drop 1 s)

242 else take 1 s ++ stripTrailingDot (drop 1 s)

247 else take 1 s ++ stripTrailingDot (drop 1 s)

243

248

244−isReservedHighRange = \s ->

249+normalizeServerName = \s -> toLower (stripTrailingDot s)

245− isAllDigitsOrDotsUntilColonOrEnd s

246− && ((take 2 s == "22" && (take 1 (drop 2 s) == "4" || take 1 (drop 2 s) == "5" || take 1 (drop 2 s) == "6" || take 1 (drop 2 s) == "7" || take 1 (drop 2 s) == "8" || take 1 (drop 2 s) == "9") && take 1 (drop 3 s) == ".")

247− || (take 2 s == "23" && isDigitChar (take 1 (drop 2 s)) && take 1 (drop 3 s) == ".")

248− || (take 2 s == "24" && isDigitChar (take 1 (drop 2 s)) && take 1 (drop 3 s) == ".")

249− || (take 3 s == "250" && take 1 (drop 3 s) == ".")

250− || (take 3 s == "251" && take 1 (drop 3 s) == ".")

251− || (take 3 s == "252" && take 1 (drop 3 s) == ".")

252− || (take 3 s == "253" && take 1 (drop 3 s) == ".")

253− || (take 3 s == "254" && take 1 (drop 3 s) == ".")

254− || (take 3 s == "255" && take 1 (drop 3 s) == "."))

255

250

256containsLocalhostSuffix = \s ->

251containsLocalhostSuffix = \s ->

257 if s == "" then False {}

252 if s == "" then False {}

@@ -278,7 +273,6 @@ isBadServerName = \s ->

278 || take 10 sn == "192.88.99."

273 || take 10 sn == "192.88.99."

279 || take 12 sn == "198.51.100."

274 || take 12 sn == "198.51.100."

280 || take 11 sn == "203.0.113."

275 || take 11 sn == "203.0.113."

281− || isReservedHighRange sn

282 || take 2 sn == "::"

276 || take 2 sn == "::"

283 || isCgnatRange sn

277 || isCgnatRange sn

284 || containsMultipleColons sn

278 || containsMultipleColons sn

@@ -319,7 +313,8 @@ safeCompareLoop = \s1 s2 result ->

319 then result + (if s1 == "" then 0 else 1)

313 then result + (if s1 == "" then 0 else 1)

320 else safeCompareLoop (drop 1 s1) (drop 1 s2) (result + (if take 1 s1 == take 1 s2 then 0 else 1))

314 else safeCompareLoop (drop 1 s1) (drop 1 s2) (result + (if take 1 s1 == take 1 s2 then 0 else 1))

321

315

322−safeCompare = \a b -> not (a == "") && not (b == "") && safeCompareLoop a b 0 == 0

316+safeCompare = \a b ->

317+ safeCompareLoop a b 0 + (if a == "" then 1 else 0) + (if b == "" then 1 else 0) == 0

323

318

324sendToEach = \servers action ->

319sendToEach = \servers action ->

325 if count servers == 0

320 if count servers == 0

@@ -409,6 +404,22 @@ broadcastOfflineGossip = \t peerServers expiredLocal -> do

409 sendBatchedGossip peerServers gossipEvents

404 sendBatchedGossip peerServers gossipEvents

410 else yield {}

405 else yield {}

411

406

407+retryPendingForwards = \pending ->

408+ if count pending == 0

409+ then yield {}

410+ else do

411+ let f = fold (\acc x -> if acc.messageId == "" then x else acc) {messageId: "", toServer: "", toKey: "", fromKey: "", encryptedBlob: "", timestamp: 0} pending

412+ let rest = do x <- pending; where not (x.messageId == f.messageId && x.toServer == f.toServer); yield x

413+ if f.messageId == ""

414+ then yield {}

415+ else do

416+ catch (fetch (federationScheme ++ f.toServer) (ForwardMessage {

417+ toKey: f.toKey, fromKey: f.fromKey, encryptedBlob: f.encryptedBlob,

418+ timestamp: f.timestamp, originServer: serverHost, messageId: f.messageId,

419+ authorization: "Bearer " ++ federationSecret})) (\err ->

420+ println ("prune: retry forward to " ++ f.toServer ++ " failed: " ++ show err))

421+ retryPendingForwards rest

422+

412-- ============================================

423-- ============================================

413-- Request Handler

424-- Request Handler

414-- ============================================

425-- ============================================

@@ -427,7 +438,7 @@ serve = \req -> case req of

427 challenges <- *challenges

438 challenges <- *challenges

428 let existingCount = count (do c <- challenges; where c.pubkey == pubkey; yield c)

439 let existingCount = count (do c <- challenges; where c.pubkey == pubkey; yield c)

429 let oldestChallengeExpiry = fold (\acc c2 -> if c2.pubkey == pubkey && (acc == -1 || c2.expiresAt < acc) then c2.expiresAt else acc) -1 challenges

440 let oldestChallengeExpiry = fold (\acc c2 -> if c2.pubkey == pubkey && (acc == -1 || c2.expiresAt < acc) then c2.expiresAt else acc) -1 challenges

430− let cappedChallenges = if existingCount >= 5 && oldestChallengeExpiry >= 0

441+ let cappedChallenges = if existingCount >= maxChallengesPerKey && oldestChallengeExpiry >= 0

431 then fold (\acc c ->

442 then fold (\acc c ->

432 if not acc.removed && c.pubkey == pubkey && c.expiresAt == oldestChallengeExpiry

443 if not acc.removed && c.pubkey == pubkey && c.expiresAt == oldestChallengeExpiry

433 then {removed: True {}, items: acc.items}

444 then {removed: True {}, items: acc.items}

@@ -444,7 +455,7 @@ serve = \req -> case req of

444 else cappedChallenges.items

455 else cappedChallenges.items

445 set *challenges = union

456 set *challenges = union

446 globalCapped

457 globalCapped

447− [{pubkey: pubkey, challenge: challenge, expiresAt: t + 60000}]

458+ [{pubkey: pubkey, challenge: challenge, expiresAt: t + challengeExpiry}]

448 yield pr

459 yield pr

449 respond {challenge: challenge}

460 respond {challenge: challenge}

450 broadcastOfflineGossip t pruneResult.peerServers pruneResult.expiredLocal

461 broadcastOfflineGossip t pruneResult.peerServers pruneResult.expiredLocal

@@ -454,7 +465,7 @@ serve = \req -> case req of

454 let pubkey = if isValidHex pubkey then normalizeHex pubkey else pubkey

465 let pubkey = if isValidHex pubkey then normalizeHex pubkey else pubkey

455 t <- now

466 t <- now

456 token <- genToken

467 token <- genToken

457− let expiry = t + 3600000

468+ let expiry = t + sessionExpiry

458 result <- atomic do

469 result <- atomic do

459 pr <- pruneExpired t

470 pr <- pruneExpired t

460 challenges <- *challenges

471 challenges <- *challenges

@@ -489,7 +500,7 @@ serve = \req -> case req of

489 if not acc.removed && s.pubkey == pubkey && s.expiresAt == oldestExpiry

500 if not acc.removed && s.pubkey == pubkey && s.expiresAt == oldestExpiry

490 then {removed: True {}, sessions: acc.sessions}

501 then {removed: True {}, sessions: acc.sessions}

491 else {removed: acc.removed, sessions: union acc.sessions [s]}) {removed: False {}, sessions: []} sessions

502 else {removed: acc.removed, sessions: union acc.sessions [s]}) {removed: False {}, sessions: []} sessions

492− let sessionsCapped = if count existingForKey >= 10 && oldestExpiry >= 0

503+ let sessionsCapped = if count existingForKey >= maxSessionsPerKey && oldestExpiry >= 0

493 then cappedFold.sessions

504 then cappedFold.sessions

494 else sessions

505 else sessions

495 let globalOldestSessionExpiry = if count sessionsCapped >= maxSessions

506 let globalOldestSessionExpiry = if count sessionsCapped >= maxSessions

@@ -533,7 +544,7 @@ serve = \req -> case req of

533 then sendToEach result.peerServers (\server ->

544 then sendToEach result.peerServers (\server ->

534 fetch (federationScheme ++ server) (RecvGossip {

545 fetch (federationScheme ++ server) (RecvGossip {

535 authorization: "Bearer " ++ federationSecret,

546 authorization: "Bearer " ++ federationSecret,

536− events: [{eventType: "online", pubkey: pubkey, server: serverHost, timestamp: t, ttl: 3600}], hopsRemaining: 2, fromServer: serverHost}))

547+ events: [{eventType: "online", pubkey: pubkey, server: serverHost, timestamp: t, ttl: onlineGossipTtl}], hopsRemaining: 2, fromServer: serverHost}))

537 else yield {}

548 else yield {}

538 broadcastOfflineGossip t result.offlinePeers (do

549 broadcastOfflineGossip t result.offlinePeers (do

539 p <- result.expiredLocal

550 p <- result.expiredLocal

@@ -638,20 +649,20 @@ serve = \req -> case req of

638 else "queued"

649 else "queued"

639 forwards <- *forwards

650 forwards <- *forwards

640 let forwardsAtCapacity = count forwards >= maxForwardsEntries

651 let forwardsAtCapacity = count forwards >= maxForwardsEntries

641− set *forwards = if shouldInsert && not forwardsAtCapacity && not localOnline

652+ set *forwards = if shouldInsert && not forwardsAtCapacity && count remoteServers > 0

642 then union forwards (do s <- remoteServers; yield {messageId: msgId, toServer: s, createdAt: t})

653 then union forwards (do s <- remoteServers; yield {messageId: msgId, toServer: s, createdAt: t})

643 else forwards

654 else forwards

644 yield {status: status, remoteServers: remoteServers, fromKey: fromKey, expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, forwardsAtCapacity: forwardsAtCapacity}

655 yield {status: status, remoteServers: remoteServers, fromKey: fromKey, expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, forwardsAtCapacity: forwardsAtCapacity}

645 let finalStatus = if result.status == "federated" && federationSecret == "" then "queued" else result.status

656 let finalStatus = if result.status == "federated" && federationSecret == "" then "queued" else result.status

646 respond {status: finalStatus, messageId: (if result.status == "unauthorized" then "" else if result.status == "invalid" then "" else msgId)}

657 respond {status: finalStatus, messageId: (if result.status == "unauthorized" then "" else if result.status == "invalid" then "" else msgId)}

647− if result.status == "federated" && not (federationSecret == "")

658+ if count result.remoteServers > 0 && not (result.status == "unauthorized") && not (result.status == "invalid") && not (federationSecret == "")

648 then sendToEach result.remoteServers (\server ->

659 then sendToEach result.remoteServers (\server ->

649 fetch (federationScheme ++ server) (ForwardMessage {

660 fetch (federationScheme ++ server) (ForwardMessage {

650 toKey: to, fromKey: result.fromKey, encryptedBlob: encryptedBlob,

661 toKey: to, fromKey: result.fromKey, encryptedBlob: encryptedBlob,

651 timestamp: clampedTs, originServer: serverHost, messageId: msgId,

662 timestamp: clampedTs, originServer: serverHost, messageId: msgId,

652 authorization: "Bearer " ++ federationSecret}))

663 authorization: "Bearer " ++ federationSecret}))

653 else yield {}

664 else yield {}

654− if result.forwardsAtCapacity && result.status == "federated"

665+ if result.forwardsAtCapacity && count result.remoteServers > 0 && not (result.status == "unauthorized") && not (result.status == "invalid")

655 then println ("sendMessage: forwards at capacity (" ++ show maxForwardsEntries ++ "), skipping forward tracking")

666 then println ("sendMessage: forwards at capacity (" ++ show maxForwardsEntries ++ "), skipping forward tracking")

656 else yield {}

667 else yield {}

657 broadcastOfflineGossip t result.peerServers result.expiredLocal

668 broadcastOfflineGossip t result.peerServers result.expiredLocal

@@ -739,6 +750,7 @@ serve = \req -> case req of

739 t <- now

750 t <- now

740 let authorized = verifyFederationAuth authorization

751 let authorized = verifyFederationAuth authorization

741 let cappedHops = if hopsRemaining > 5 then 5 else hopsRemaining

752 let cappedHops = if hopsRemaining > 5 then 5 else hopsRemaining

753+ let fromServer = normalizeServerName fromServer

742 let validRequest = authorized && cappedHops >= 1 && not (fromServer == "") && not (fromServer == serverHost) && not (tooLong fromServer maxServerNameLen) && not (isBadServerName fromServer) && count events <= maxGossipEvents

754 let validRequest = authorized && cappedHops >= 1 && not (fromServer == "") && not (fromServer == serverHost) && not (tooLong fromServer maxServerNameLen) && not (isBadServerName fromServer) && count events <= maxGossipEvents

743 let onlineEntries = do

755 let onlineEntries = do

744 e <- events

756 e <- events

@@ -750,10 +762,10 @@ serve = \req -> case req of

750 where not (tooLong e.server maxServerNameLen)

762 where not (tooLong e.server maxServerNameLen)

751 where not (isBadServerName e.server)

763 where not (isBadServerName e.server)

752 where isValidHex e.pubkey

764 where isValidHex e.pubkey

753− where not (e.server == serverHost)

765+ where not (normalizeServerName e.server == serverHost)

754 where e.timestamp <= t + clockSkewTolerance

766 where e.timestamp <= t + clockSkewTolerance

755 where e.timestamp + (clampTtl e.ttl) * 1000 > t

767 where e.timestamp + (clampTtl e.ttl) * 1000 > t

756− yield {pubkey: normalizeHex e.pubkey, server: e.server, expiresAt: e.timestamp + (clampTtl e.ttl) * 1000, timestamp: e.timestamp}

768+ yield {pubkey: normalizeHex e.pubkey, server: normalizeServerName e.server, expiresAt: e.timestamp + (clampTtl e.ttl) * 1000, timestamp: e.timestamp}

757 let offlineEntries = do

769 let offlineEntries = do

758 e <- events

770 e <- events

759 where validRequest

771 where validRequest

@@ -764,10 +776,10 @@ serve = \req -> case req of

764 where not (tooLong e.server maxServerNameLen)

776 where not (tooLong e.server maxServerNameLen)

765 where not (isBadServerName e.server)

777 where not (isBadServerName e.server)

766 where isValidHex e.pubkey

778 where isValidHex e.pubkey

767− where not (e.server == serverHost)

779+ where not (normalizeServerName e.server == serverHost)

768 where e.timestamp <= t + clockSkewTolerance

780 where e.timestamp <= t + clockSkewTolerance

769 where t - e.timestamp < (clampTtl e.ttl) * 1000

781 where t - e.timestamp < (clampTtl e.ttl) * 1000

770− yield {pubkey: normalizeHex e.pubkey, server: e.server, timestamp: e.timestamp}

782+ yield {pubkey: normalizeHex e.pubkey, server: normalizeServerName e.server, timestamp: e.timestamp}

771 let filteredOnline = do

783 let filteredOnline = do

772 e <- onlineEntries

784 e <- onlineEntries

773 where count (do

785 where count (do

@@ -941,20 +953,20 @@ serve = \req -> case req of

941 where not (tooLong e.server maxServerNameLen)

953 where not (tooLong e.server maxServerNameLen)

942 where not (isBadServerName e.server)

954 where not (isBadServerName e.server)

943 where isValidHex e.pubkey

955 where isValidHex e.pubkey

944− where not (e.server == serverHost)

956+ where not (normalizeServerName e.server == serverHost)

945 where e.timestamp <= t + clockSkewTolerance

957 where e.timestamp <= t + clockSkewTolerance

946 where e.timestamp + (clampTtl e.ttl) * 1000 > t

958 where e.timestamp + (clampTtl e.ttl) * 1000 > t

947 where not (e.eventType == "online" && count (do

959 where not (e.eventType == "online" && count (do

948 o <- offlineEntries

960 o <- offlineEntries

949 where o.pubkey == normalizeHex e.pubkey

961 where o.pubkey == normalizeHex e.pubkey

950− where o.server == e.server

962+ where o.server == normalizeServerName e.server

951 where o.timestamp > e.timestamp

963 where o.timestamp > e.timestamp

952 yield o) > 0)

964 yield o) > 0)

953 yield e

965 yield e

954 let effectiveEvents = do

966 let effectiveEvents = do

955 e <- validEvents

967 e <- validEvents

956− where count (do a <- gossipResult.appliedKeys; where a.pubkey == normalizeHex e.pubkey; where a.server == e.server; yield a) > 0

968+ where count (do a <- gossipResult.appliedKeys; where a.pubkey == normalizeHex e.pubkey; where a.server == normalizeServerName e.server; yield a) > 0

957− yield {eventType: e.eventType, pubkey: normalizeHex e.pubkey, server: e.server, timestamp: e.timestamp, ttl: e.ttl}

969+ yield {eventType: e.eventType, pubkey: normalizeHex e.pubkey, server: normalizeServerName e.server, timestamp: e.timestamp, ttl: e.ttl}

958 respond {ok: validRequest}

970 respond {ok: validRequest}

959 if not gossipResult.withinCapacity

971 if not gossipResult.withinCapacity

960 then println ("gossip: presence at capacity (" ++ show maxPresenceEntries ++ "), limiting new entries from " ++ fromServer)

972 then println ("gossip: presence at capacity (" ++ show maxPresenceEntries ++ "), limiting new entries from " ++ fromServer)

@@ -981,6 +993,7 @@ serve = \req -> case req of

981 let fromKey = if isValidHex fromKey then normalizeHex fromKey else fromKey

993 let fromKey = if isValidHex fromKey then normalizeHex fromKey else fromKey

982 let encryptedBlob = if isValidHex encryptedBlob then normalizeHex encryptedBlob else encryptedBlob

994 let encryptedBlob = if isValidHex encryptedBlob then normalizeHex encryptedBlob else encryptedBlob

983 let messageId = if isValidHex messageId then normalizeHex messageId else messageId

995 let messageId = if isValidHex messageId then normalizeHex messageId else messageId

996+ let originServer = normalizeServerName originServer

984 t <- now

997 t <- now

985 let authorized = verifyFederationAuth authorization

998 let authorized = verifyFederationAuth authorization

986 let validOrigin = not (originServer == "") && not (originServer == serverHost) && not (tooLong originServer maxServerNameLen) && not (isBadServerName originServer)

999 let validOrigin = not (originServer == "") && not (originServer == serverHost) && not (tooLong originServer maxServerNameLen) && not (isBadServerName originServer)

@@ -1013,6 +1026,8 @@ serve = \req -> case req of

1013 broadcastOfflineGossip t result.peerServers result.expiredLocal

1026 broadcastOfflineGossip t result.peerServers result.expiredLocal

1014

1027

1015 FederationAck {messageIds, originServer, fromServer, authorization, respond} -> do

1028 FederationAck {messageIds, originServer, fromServer, authorization, respond} -> do

1029+ let fromServer = normalizeServerName fromServer

1030+ let originServer = normalizeServerName originServer

1016 t <- now

1031 t <- now

1017 let authorized = verifyFederationAuth authorization

1032 let authorized = verifyFederationAuth authorization

1018 let dedupedIds = fold (\acc mid ->

1033 let dedupedIds = fold (\acc mid ->

@@ -1027,13 +1042,14 @@ serve = \req -> case req of

1027 pr <- pruneExpired t

1042 pr <- pruneExpired t

1028 forwards <- *forwards

1043 forwards <- *forwards

1029 let validAck = validAckFields

1044 let validAck = validAckFields

1045+ let forwardsToRemove = do

1046+ f <- forwards

1047+ mid <- dedupedIds

1048+ where f.messageId == mid

1049+ where f.toServer == fromServer

1050+ yield f

1030 set *forwards = if validAck

1051 set *forwards = if validAck

1031− then diff forwards (do

1052+ then diff forwards forwardsToRemove

1032− f <- forwards

1033− mid <- dedupedIds

1034− where f.messageId == mid

1035− where f.toServer == fromServer

1036− yield f)

1037 else forwards

1053 else forwards

1038 yield {peerServers: pr.peerServers, expiredLocal: pr.expiredLocal, validAck: validAck}

1054 yield {peerServers: pr.peerServers, expiredLocal: pr.expiredLocal, validAck: validAck}

1039 respond {ok: pruneResult.validAck}

1055 respond {ok: pruneResult.validAck}

@@ -1046,9 +1062,20 @@ serve = \req -> case req of

1046 then do

1062 then do

1047 result <- atomic do

1063 result <- atomic do

1048 pr <- pruneExpired t

1064 pr <- pruneExpired t

1049− yield pr

1065+ forwards <- *forwards

1066+ messages <- *messages

1067+ let pendingForwards = do

1068+ f <- forwards

1069+ m <- messages

1070+ where m.messageId == f.messageId

1071+ where m.originServer == ""

1072+ yield {toServer: f.toServer, toKey: m.toKey, fromKey: m.fromKey, encryptedBlob: m.encryptedBlob, timestamp: m.timestamp, messageId: m.messageId}

1073+ yield {expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, pendingForwards: pendingForwards}

1050 respond {ok: True {}}

1074 respond {ok: True {}}

1051 broadcastOfflineGossip t result.peerServers result.expiredLocal

1075 broadcastOfflineGossip t result.peerServers result.expiredLocal

1076+ if not (federationSecret == "") && count result.pendingForwards > 0

1077+ then retryPendingForwards result.pendingForwards

1078+ else yield {}

1052 else respond {ok: False {}}

1079 else respond {ok: False {}}

1053

1080

1054-- ============================================

1081-- ============================================