diff --git a/Tests/Host.lean b/Tests/Host.lean index 7d444cc..662dd86 100644 --- a/Tests/Host.lean +++ b/Tests/Host.lean @@ -3,6 +3,11 @@ import Tests.Support import Conc import Streams import Multiformats +import Libp2p.Identity +import Libp2p.Core +import Libp2p.Security +import Libp2p.Mux +import Libp2p.Negotiate import Libp2p.Host /-! @@ -16,13 +21,11 @@ pure update functions, by value or under an `OnceRef`, with no connection in the loop. The session handle of a table entry is a `Nat` here, since the rule never reads it. -Unit 1b runs hosts over the sim transport, with the real -`Noise.securityProtocol` and `Mux.yamux` values (PRINCIPLES 1.4). Every -host derives its peer id through `libp2p/identity`, and the upgrade runs -through `libp2p/negotiate`. The real packages have landed (#53, #76, -#77), and the upgrade still runs through the negotiate mock of the -host, whose entries fail. The legs wait for the host swap of issue -#70. Until then the suite names each leg as a skip. +Unit 1b: hosts over the test values of root §3.1 (PRINCIPLES 1.4): the +sim transport, the fixed-key test value `Noise.fixedKeyProtocol`, and +a test clock that drives the upgrade deadline, so no check waits on the +wall clock. The muxer is `Mux.yamux`. Each host hands its failures to a +channel that the check reads. -/ set_option autoImplicit false @@ -278,18 +281,360 @@ private def errorChecks (r : Runner) : IO Unit := do for (e, line) in arms do r.check s!"Host.Error renders {line}" (toString e == line) -/-! ## Unit 1b -/ - -/-- The unit 1b legs of the plan. Each one needs a host, and a host -derives its peer id through `libp2p/identity` and upgrades through -`libp2p/negotiate` (module doc). -/ -private def swarmLegs : List String := - ["the upgrade pipeline in both directions, with the identity check", - "the host's own answers: dial to self, no address, no transport, unknown protocol, failing handler, after close", - "the policy doubles: a denying gater refuses a dial and an inbound connection", - "the silent peer at the deadline, and close reaching a parked upgrade", - "the shutdown stages of §5.5 in order", - "the leftover hand-over on a served stream"] +/-! ## Unit 1b: hosts over the sim -/ + +section Swarm + +open Streams Libp2p.Security +open Libp2p.Host (Host Config Failure Stage) + +/-- `n` bytes of the value `v`. -/ +private def fill (n : Nat) (v : UInt8) : ByteArray := ⟨Array.replicate n v⟩ + +/-- A Noise configuration from an Ed25519 identity byte. The static key +byte follows from it. -/ +private def cfgOf (idByte : UInt8) : Async Noise.Config := do + let .ok identity := PrivKey.ofData? .ed25519 (fill 32 idByte) + | throw <| IO.userError "test identity does not build" + let some static := Noise.Bytes32.ofByteArray? (fill 32 (idByte + 100)) + | throw <| IO.userError "test static key is not 32 bytes" + return { identity, static } + +/-- The fixed-key Noise value of the identity byte `idByte`: its +ephemeral keys come from an `Rng` seeded with 32 copies of +`idByte + 50` (root §3.1). -/ +private def noiseOf (idByte : UInt8) : Async SecurityProtocol := + return Noise.fixedKeyProtocol (← cfgOf idByte) (fill 32 (idByte + 50)) + +/-- The echo protocol of the suite. The `/test/` prefix keeps it apart +from every real id. -/ +private def echoId : ProtocolId := ⟨"/test/echo"⟩ + +/-- A protocol whose handler fails at once. -/ +private def failId : ProtocolId := ⟨"/test/fail"⟩ + +/-- An id no host serves. -/ +private def unknownId : ProtocolId := ⟨"/test/unknown"⟩ + +/-- Read until the end of the stream, or a reset. -/ +private def readAll (s : Stream) : Async ByteArray := do + let mut acc := ByteArray.empty + repeat + match ← s.read with + | .ok (some b) => acc := acc ++ b + | _ => break + return acc + +/-- The echo handler: read to the end, write it back, close. -/ +private def echoHandler : StreamHandler where + id := echoId + handle _ s := do + let got ← readAll s + discard <| s.write got + s.close + +/-- The id of the handler that records its context. -/ +private def ctxId : ProtocolId := ⟨"/test/ctx"⟩ + +/-- The failing handler. -/ +private def failHandler : StreamHandler where + id := failId + handle _ _ := throw (.protocol "the test handler fails") + +/-- A host of the suite on `net`, its peer id, and the channel its +failures arrive on. -/ +private def mkHost (net : Sim.Net) (idByte : UInt8) (tc : TestClock) + (gater : ConnectionGater := Policy.permissiveGater) : + Async (Host × Chan Failure) := do + let cfg ← cfgOf idByte + let failures : Chan Failure ← Chan.new 16 + let config : Config := { + identity := cfg.identity + transports := [net.transport] + security := [← noiseOf idByte] + muxers := [Libp2p.Mux.yamux] + handlers := [echoHandler, failHandler] + gater + clock := tc.clock + onFailure := fun f => discard (failures.trySend f) } + match ← (Host.new config : BaseIO _) with + | .ok h => return (h, failures) + | .error _ => throw <| IO.userError "test host does not build" + +/-- The address with `/p2p/` after it. -/ +private def withPeer (a : Multiaddr) (p : PeerId) : Async Multiaddr := do + match a.encapsulate (← ma s!"/p2p/{p}") with + | .ok full => return full + | .error _ => throw <| IO.userError "the /p2p address does not build" + +/-- Start a listener, or fail the run. -/ +private def listenOn (h : Host) (text : String) : Async Multiaddr := do + match ← h.startListening (← ma text) with + | .ok a => return a + | .error e => throw <| IO.userError s!"listen failed: {e}" + +/-- One echo round trip from `h` to `peer`: the stream negotiates +`echoId`, the bytes come back, and the stream ends. -/ +private def echoOnce (h : Host) (peer : PeerId) (payload : String) : Async Bool := do + match ← h.newStream peer [echoId] with + | .error _ => return false + | .ok (id, s) => + discard <| s.write payload.toUTF8 + s.close + return id == echoId && (← readAll s) == payload.toUTF8 + +/-- The pipeline in both directions: a dial with `/p2p`, an echo each +way, a redial that reuses the session, and an identity mismatch. -/ +private def pipelineChecks (r : Runner) : Async Unit := do + let net ← Sim.Net.new + let tc ← TestClock.new + let (a, _) ← mkHost net 1 tc + let (b, _) ← mkHost net 2 tc + let addrA ← listenOn a "/memory/101" + let addrB ← listenOn b "/memory/102" + -- B dials A by its address and peer id. + match ← b.dial (← withPeer addrA a.localPeer) with + | .ok p => r.check "host: a dial proves the listener's peer id" (p == a.localPeer) + | .error e => r.check s!"host: a dial succeeds ({e})" false + r.check "host: the dialer serves the session" (← (b.connected a.localPeer : BaseIO Bool)) + r.check "host: the dialer's stream echoes" (← echoOnce b a.localPeer "ping") + -- The listener committed the session too, and it opens streams back. + r.check "host: the listener's stream echoes" (← echoOnce a b.localPeer "pong") + -- A served stream gets the session's context: the proved peer, and + -- the connection's remote address. B dialed, so B's connection names + -- `addrA` as its remote end. A's connection names the sim port of + -- B's dial. + let seenA : Chan StreamContext ← Chan.new 1 + let seenB : Chan StreamContext ← Chan.new 1 + let recorder (seen : Chan StreamContext) : StreamHandler := + { id := ctxId, handle := fun ctx s => do discard <| seen.send ctx; s.close } + a.setHandler (recorder seenA) + b.setHandler (recorder seenB) + discard <| b.newStream a.localPeer [ctxId] + discard <| a.newStream b.localPeer [ctxId] + match ← seenA.recv, ← seenB.recv with + | some ca, some cb => + r.check "host: the handler context names the proved peer" + (ca.remotePeer == b.localPeer && cb.remotePeer == a.localPeer) + r.check "host: the dialer's handler context names the listener's address" + (cb.remoteAddr == addrA && ca.remoteAddr != addrA) + | _, _ => r.check "host: both handlers saw a context" false + -- A redial reuses the live session, and the peerstore grows nothing. + match ← b.dial (← withPeer addrA a.localPeer) with + | .ok p => r.check "host: a redial reuses the session" (p == a.localPeer) + | .error e => r.check s!"host: a redial succeeds ({e})" false + r.check "host: the dial stored the address once" + ((← (b.addrsOf a.localPeer : BaseIO (List Multiaddr))) == [addrA]) + -- A dial that expects another peer fails at the identity check. + let (c, _) ← mkHost net 3 tc + match ← c.dial addrB (peer := some a.localPeer) with + | .error (.upgrade (.authentication _)) => + r.check "host: an identity mismatch fails with authentication" true + | _ => r.check "host: an identity mismatch fails with authentication" false + r.check "host: the mismatched dial leaves no session" !(← (c.connected b.localPeer : BaseIO Bool)) + c.close + b.close + a.close + tc.close + +/-- The host's own answers: a dial to itself, a peer with no address, +an address with no transport part, an unknown protocol, a failing +handler, and the calls after `close`. -/ +private def answerChecks (r : Runner) : Async Unit := do + let net ← Sim.Net.new + let tc ← TestClock.new + let (a, failsA) ← mkHost net 4 tc + let (b, _) ← mkHost net 5 tc + let addrA ← listenOn a "/memory/111" + r.check "host: a dial to our own id fails" + ((← b.dial addrA (peer := some b.localPeer)) matches .error .dialSelf) + r.check "host: a peer with no address fails" + ((← b.dialPeer a.localPeer) matches .error (.noAddresses _)) + r.check "host: a /p2p address alone has no transport" + ((← b.dial (← ma s!"/p2p/{a.localPeer}")) matches .error (.noTransport _)) + -- `dialPeer` walks the peerstore. + b.addAddrs a.localPeer [addrA] + r.check "host: dialPeer dials the peerstore address" + ((← b.dialPeer a.localPeer) matches .ok _) + -- The host answers `na` to an unknown protocol, and the session stays. + r.check "host: an unknown protocol fails the stream" + ((← b.newStream a.localPeer [unknownId]) matches .error (.stream _)) + r.check "host: the session stays after an unknown protocol" + (← echoOnce b a.localPeer "still here") + -- A handler set on a running host serves the next stream. + (a.setHandler { echoHandler with id := unknownId } : BaseIO Unit) + r.check "host: a handler set at runtime serves the next stream" + ((← b.newStream a.localPeer [unknownId]) matches .ok _) + -- A failing handler resets the stream and reaches A's owner. The + -- session stays. A sends its reset right after its negotiation reply, + -- and a reset may drop the bytes in flight (`Streams.Error.reset`). + -- So B sees the reset in its first read. Or the reset comes before B + -- reads the reply, and B's negotiation ends with `closed`. + let resetSeen ← match ← b.newStream a.localPeer [failId] with + | .error (.stream (.closed _)) => pure true + | .error _ => pure false + | .ok (_, s) => pure ((← s.read) matches .error .reset) + r.check "host: a handler failure resets the stream" resetSeen + match ← failsA.recv with + | some f => + r.check "host: a handler failure reaches the owner" + (f.peer == some b.localPeer && f.protocol == some failId && + f.error matches .stream (.protocol _)) + | none => r.check "host: a handler failure reaches the owner" false + r.check "host: the session stays after a handler failure" + (← echoOnce b a.localPeer "and here") + b.close + a.close + r.check "host: a dial after close fails with closing" + ((← b.dial addrA) matches .error .closing) + r.check "host: a listen after close fails with closing" + ((← a.startListening (← ma "/memory/112")) matches .error .closing) + tc.close + +/-- A gater that passes the dial and the accept, and refuses every +proved peer: only the secured checkpoint answers no. -/ +private def securedOnlyGater : ConnectionGater := + { Policy.permissiveGater with allowSecured := fun _ _ _ _ => pure false } + +/-- The policy doubles. A denying gater refuses a dial and an accepted +connection. The permissive gater passes both, in the other checks. A +gater that refuses at the secured checkpoint refuses after the +handshake, and the scope and the secure connection close with it. -/ +private def policyChecks (r : Runner) : Async Unit := do + let net ← Sim.Net.new + let tc ← TestClock.new + let (open_, _) ← mkHost net 6 tc + let (shut, failsShut) ← mkHost net 7 tc (gater := Policy.denyingGater) + let addrOpen ← listenOn open_ "/memory/121" + let addrShut ← listenOn shut "/memory/122" + r.check "host: a denying gater refuses a dial" + ((← shut.dial addrOpen) matches .error (.gated .dial)) + -- The refused inbound connection closes before any byte moves. The + -- dialer meets the close as the end of its read (`closed`), or as a + -- failed write, which is a fault (`Connection.write`, §3.5). The + -- order of its write and the close decides which. + let refused ← try + pure ((← open_.dial addrShut) matches .error (.upgrade (.closed _))) + catch _ => pure true + r.check "host: a denying gater refuses an inbound connection" refused + r.check "host: the refusal is no failure" (← (failsShut.tryRecv : BaseIO _)).isNone + r.check "host: the refused inbound peer has no session" + !(← (shut.connected open_.localPeer : BaseIO Bool)) + -- The secured checkpoint: the dial and the handshake pass, and the + -- checkpoint refuses the proved peer before the muxer stage. + let (picky, failsPicky) ← mkHost net 13 tc (gater := securedOnlyGater) + r.check "host: the secured checkpoint refuses an outbound peer" + ((← picky.dial addrOpen) matches .error (.gated .secured)) + r.check "host: the outbound secured refusal leaves no session" + !(← (picky.connected open_.localPeer : BaseIO Bool)) + let addrPicky ← listenOn picky "/memory/123" + -- The dialer's handshake completes, then its muxer stage meets the + -- close. A failed write is a fault (§3.5), as above. + let refusedIn ← try + pure ((← open_.dial addrPicky) matches .error (.upgrade _)) + catch _ => pure true + r.check "host: the secured checkpoint refuses an inbound peer" refusedIn + r.check "host: the secured refusal is no failure" (← (failsPicky.tryRecv : BaseIO _)).isNone + r.check "host: the inbound secured refusal leaves no session" + !(← (picky.connected open_.localPeer : BaseIO Bool)) + picky.close + open_.close + shut.close + tc.close + +/-- True when `raw` reaches its end within eight reads. The host's +responder writes its multistream header before it reads, as go's +listener does (`Negotiate.negotiate`). So a raw peer can read that +header before the end. A peer that sends past the bound counts as no +end. -/ +private def endsSoon (raw : Connection) : Async Bool := do + for _ in [0:8] do + if (← raw.read).isNone then return true + return false + +/-- The silent peer, at host level (PRINCIPLES 5.4): a raw connection +that sends nothing parks the inbound upgrade. The deadline fails it, +and the failure reaches the owner. In a second run, `close` reaches +the parked upgrade through the pending list, and every join returns. -/ +private def silentPeerChecks (r : Runner) : Async Unit := do + let net ← Sim.Net.new + let tc ← TestClock.new + let (a, failsA) ← mkHost net 8 tc + let addrA ← listenOn a "/memory/131" + let some raw ← net.transport.dial addrA | throw <| IO.userError "raw dial failed" + -- The upgrade arms its deadline before its first read parks. + discard tc.nextTimer + tc.advance Streams.unauthDeadlineMs + match ← failsA.recv with + | some f => + r.check "host: a silent peer fails at the deadline" + (f.peer.isNone && f.error matches .upgrade (.timeout _)) + | none => r.check "host: a silent peer fails at the deadline" false + r.check "host: the deadline closed the silent connection" (← endsSoon raw) + -- A second silent peer, and `close` before the deadline. + let some raw ← net.transport.dial addrA | throw <| IO.userError "raw dial failed" + discard tc.nextTimer + a.close + r.check "host: close reaches a parked upgrade, and returns" (← endsSoon raw) + tc.close + +/-- The shutdown order (§5.5): the stages run in sequence, and every +join returns, with a listener whose accept parks, a live session, and +a pending connection. -/ +private def shutdownChecks (r : Runner) : Async Unit := do + let net ← Sim.Net.new + let tc ← TestClock.new + let (a, _) ← mkHost net 9 tc + let (b, _) ← mkHost net 10 tc + let addrA ← listenOn a "/memory/141" + discard <| listenOn a "/memory/142" + discard <| b.dial addrA + let some raw ← net.transport.dial addrA | throw <| IO.userError "raw dial failed" + let stages : OnceRef (List Stage) ← OnceRef.new [] + a.close (onStage := fun s => stages.update (· ++ [s])) + r.check "host: the shutdown runs the stages of §5.5 in order" + ((← (stages.get : BaseIO (List Stage))) == + [.flag, .listeners, .pending, .sessions, .sessionLoops, .hostTasks]) + r.check "host: the shutdown closed the pending connection" (← endsSoon raw) + r.check "host: a closed listener refuses a dial" + ((← b.dial addrA (peer := none)) matches .error _) + -- A second close finds nothing left, and returns. + a.close + b.close + tc.close + +/-- The leftover hand-over on a served stream (§4.7, §5.6). A peer +writes its stream header, its proposal, and data in one write. The +handler gets the data in its first read. The peer runs the plain +upgrader of `libp2p/negotiate`, so the host's listen path meets a peer +that is no host. -/ +private def handOverChecks (r : Runner) : Async Unit := do + let net ← Sim.Net.new + let tc ← TestClock.new + let (a, _) ← mkHost net 11 tc + let addrA ← listenOn a "/memory/151" + let noise ← noiseOf 12 + let some conn ← net.transport.dial addrA | throw <| IO.userError "dial failed" + match ← (Libp2p.Negotiate.upgradeWith tc.clock .initiator conn .empty (some a.localPeer) + [noise] [Libp2p.Mux.yamux]).run with + | .error e => r.check s!"host: a plain upgrader reaches the host ({e})" false + | .ok (session, _) => + match ← (session.openStream echoId).run with + | .error e => r.check s!"host: the stream opens ({e})" false + | .ok s => + let msg (t : String) : ByteArray := Bytes.ofNatList (Libp2p.Negotiate.encodeText t) + discard <| s.write (msg "/multistream/1.0.0" ++ msg "/test/echo" ++ "coalesced".toUTF8) + s.close + -- The echo handler answers the header and the id, then the bytes. + let back ← readAll s + r.check "host: the bytes behind the proposal reach the handler" + (back == msg "/multistream/1.0.0" ++ msg "/test/echo" ++ "coalesced".toUTF8) + discard session.close.toBaseIO + discard session.join.run + a.close + tc.close + +end Swarm def Host.run (r : Runner) : IO Unit := do r.section "host" "(packages/libp2p/host/docs/PLAN.md, test milestones, units 1a and 1b)" @@ -299,7 +644,10 @@ def Host.run (r : Runner) : IO Unit := do (handlerChecks r).run (peerstoreChecks r).run errorChecks r - for leg in swarmLegs do - r.skip s!"host: {leg}" "needs the negotiate swap of the host, issue #70" + for (label, act) in [("pipeline", pipelineChecks r), ("answers", answerChecks r), + ("policy", policyChecks r), ("silent peer", silentPeerChecks r), + ("shutdown", shutdownChecks r), ("hand-over", handOverChecks r)] do + try act.run + catch e => r.check s!"host {label}: runs ({e})" false end Tests \ No newline at end of file diff --git a/Tests/HostPair.lean b/Tests/HostPair.lean new file mode 100644 index 0000000..17770df --- /dev/null +++ b/Tests/HostPair.lean @@ -0,0 +1,83 @@ +import Conc +import Streams +import Libp2p.Identity +import Libp2p.Security +import Libp2p.Mux +import Libp2p.Host +import Multiformats.Multiaddr + +/-! +# Two hosts over the sim + +The protocol suites run their handlers between two real hosts, on the +test values of root §3.1 (PRINCIPLES 1.4): the in-memory sim transport, +the fixed-key Noise value with Ed25519 identities from one byte each, +and a test clock. The muxer is yamux. The server listens on +`/memory/4100`, and the client holds a session to it. +-/ + +set_option autoImplicit false + +namespace Tests.HostPair + +open Conc (Async Clock OnceRef TestClock) +open Libp2p.Host (Host Failure) + +/-- A host on `net` and `clock`: the fixed-key Noise value with an +Ed25519 identity from one byte, yamux, and the sim transport. The +ephemeral keys come from an `Rng` seeded with 32 copies of +`idByte + 50`. Its failures go to `failures`. -/ +def mkHost (net : Streams.Sim.Net) (clock : Clock) (idByte : UInt8) + (failures : OnceRef (List Failure)) : Async Host := do + let .ok identity := Libp2p.Identity.PrivKey.ofData? .ed25519 ⟨Array.replicate 32 idByte⟩ + | throw (IO.userError "test identity does not build") + let some static := Libp2p.Security.Noise.Bytes32.ofByteArray? ⟨Array.replicate 32 (idByte + 100)⟩ + | throw (IO.userError "test static key is not 32 bytes") + let noise := Libp2p.Security.Noise.fixedKeyProtocol { identity, static } + ⟨Array.replicate 32 (idByte + 50)⟩ + match ← (Host.new { + identity + transports := [net.transport] + security := [noise] + muxers := [Libp2p.Mux.yamux] + clock + onFailure := fun f => failures.update (f :: ·) } : BaseIO _) with + | .ok h => return h + | .error _ => throw (IO.userError "test host does not build") + +/-- Two hosts over a new sim net: the server listens, and the client +holds a session to it. -/ +structure Pair where + client : Host + server : Host + /-- The peer id the client proved for the server. -/ + serverPeer : Libp2p.Identity.PeerId + /-- The address the server listens on. -/ + serverAddr : Multiformats.Multiaddr + /-- The failures of the server's owned tasks, newest first. -/ + failures : OnceRef (List Failure) + /-- The test clock both hosts run on. `Pair.close` closes it. -/ + clock : TestClock + +/-- The pair. With `split`, the sim delivers each byte as its own read. -/ +def hostPair (split : Bool := false) : Async Pair := do + let net ← Streams.Sim.Net.new split + let tc ← TestClock.new + let failures ← OnceRef.new [] + let client ← mkHost net tc.clock 1 (← OnceRef.new []) + let server ← mkHost net tc.clock 2 failures + let addr ← match Multiformats.Multiaddr.parse? "/memory/4100" with + | .ok a => pure a + | .error _ => throw (IO.userError "test address") + let .ok bound ← server.startListening addr | throw (IO.userError "listen failed") + let .ok serverPeer ← client.dial bound | throw (IO.userError "dial failed") + return { client, server, serverPeer, serverAddr := bound, failures, clock := tc } + +/-- Close both hosts, then the clock. Each close joins its own tasks +(§5.5). -/ +def Pair.close (p : Pair) : Async Unit := do + p.client.close + p.server.close + p.clock.close + +end Tests.HostPair diff --git a/Tests/ReqResp.lean b/Tests/ReqResp.lean index 1970345..b386b9a 100644 --- a/Tests/ReqResp.lean +++ b/Tests/ReqResp.lean @@ -1,7 +1,7 @@ import Tests.Support +import Tests.HostPair import Libp2p.Protocols.ReqResp.Exchange -import Libp2p.Protocols.Mock.Muxer -import Libp2p.Protocols.Mock.HostStream +import Libp2p.Host import Multiformats.Multiaddr /-! @@ -18,8 +18,8 @@ encoder shape writes through go's snappy. The sections follow the layers: the checksum, the chunk codec, the length prefix, the reader core as transcripts with no clock and no task, the writer, the concurrency limit, the exchange over a stream -pair with the test clock, and the exchange over the sim transport -through the mock muxer and the host stream mock. +pair with the test clock, and the exchange between two hosts over the +sim transport, with the fixed-key Noise value and yamux. -/ set_option autoImplicit false @@ -607,35 +607,17 @@ def limiterChecks (r : Runner) : IO Unit := do /-! ## Over the sim transport -/ -open Libp2p.Protocols.Mock.Muxer (muxer) -open Libp2p.Protocols.Mock.HostStream (Host) - -/-- The peer id of a fixed Ed25519 seed of 32 copies of `v`. -/ -private def seedPeer (v : UInt8) : Async Libp2p.Identity.PeerId := do - let seed : ByteArray := ⟨Array.replicate 32 v⟩ - match (do Libp2p.Identity.PeerId.ofPubKey (← (← Libp2p.Identity.PrivKey.ofData? .ed25519 seed).toPublic) - : Except Libp2p.Identity.Error Libp2p.Identity.PeerId) with - | .ok p => return p - | .error _ => throw (IO.userError "test: the fixed seed derives no peer id") - -/-- A connected pair of hosts over a new sim net and the mock muxer. -Each end's handler context names the other end's peer and its own -connection's remote address. -/ -def hostPair (split : Bool := false) : Async (Host × Host) := do - let net ← Streams.Sim.Net.new split - let addr ← match Multiformats.Multiaddr.parse? "/ip4/127.0.0.1/tcp/4100" with - | .ok a => pure a - | .error _ => throw (IO.userError "test address") - let l ← net.transport.listen addr - let some c ← net.transport.dial l.addr | throw (IO.userError "dial failed") - let some s ← l.accept | throw (IO.userError "accept failed") - l.close - let .ok cs ← (muxer.«open» .initiator c ByteArray.empty).run | throw (IO.userError "open failed") - let .ok ss ← (muxer.«open» .responder s ByteArray.empty).run | throw (IO.userError "open failed") - let client ← seedPeer 1 - let server ← seedPeer 2 - return (← Host.new cs { remotePeer := server, remoteAddr := c.remoteAddr }, - ← Host.new ss { remotePeer := client, remoteAddr := s.remoteAddr }) +open Tests.HostPair (Pair hostPair) + +/-- The host's stream call for the server, as the `OpenStream` of +`requestPeer`. A host failure that wraps a seam error reaches the +requester as that error. The host renders its own refusals into +`protocol`. -/ +def openStreamOf (p : Pair) : OpenStream := fun ids => do + match ← p.client.newStream p.serverPeer ids with + | .ok x => return .ok x + | .error (.stream e) | .error (.upgrade e) | .error (.session e) => return .error e + | .error e => return .error (.protocol (toString e)) def simChecks (r : Runner) : Async Unit := do r.section "over the sim" "phase0 Req/Resp interaction; ARCHITECTURE §5.4; PRINCIPLES 1.4" @@ -644,83 +626,69 @@ def simChecks (r : Runner) : Async Unit := do let answer : List Chunk := [⟨.success, digest, noise 1 70000⟩, ⟨.success, digest, noise 2 5⟩] for split in [false, true] do let tc ← TestClock.new - let (client, server) ← hostPair split - server.setHandler (handler m tc.clock 1000 fun _ => return answer) - server.serve + let p ← hostPair split + p.server.setHandler (handler m tc.clock 1000 fun _ => return answer) let limiter ← Limiter.new let got : OnceRef (List Chunk) ← OnceRef.new [] - let res ← requestPeer client.newStream limiter m tc.clock 1000 (some (noise 3 200)) + let res ← requestPeer (openStreamOf p) limiter m tc.clock 1000 (some (noise 3 200)) fun c => got.update (· ++ [c]) r.check s!"sim (split {split}): a request through the host gets its chunks" (succeedsWith () res && (← got.get).map (·.payload) == answer.map (·.payload)) - r.check s!"sim (split {split}): the handler saw no failure" ((← server.failures.get).isEmpty) + r.check s!"sim (split {split}): the handler saw no failure" ((← p.failures.get).isEmpty) r.check s!"sim (split {split}): the limiter slot came back" ((← limiter.take m.id) && (← limiter.take m.id)) - client.close - server.close - client.join - server.join + p.close tc.close - -- An id the peer does not serve: the stream is reset. + -- An id the peer does not serve: multistream-select answers `na`, so + -- the stream never opens for it. do let tc ← TestClock.new - let (client, server) ← hostPair - server.serve - let res ← requestPeer client.newStream (← Limiter.new) m tc.clock 1000 (some (noise 3 4)) + let p ← hostPair + let res ← requestPeer (openStreamOf p) (← Limiter.new) m tc.clock 1000 (some (noise 3 4)) fun _ => pure () - r.check "sim: an id with no handler fails the request with a reset" - (failsWith (.stream .reset) res) - client.close - server.close - client.join - server.join + r.check "sim: an id with no handler fails the open" + (match res with | .error (.openFailed _) => true | _ => false) + p.close tc.close -- A silent peer over the sim: the deadline resets the stream, and the -- connection stays for the next exchange. do let tc ← TestClock.new - let (client, server) ← hostPair + let p ← hostPair -- The handler takes one read and ends with no answer and no close. - server.setHandler ⟨m.id, fun _ s => ExceptT.mk do + p.server.setHandler ⟨m.id, fun _ s => ExceptT.mk do discard <| s.read return .ok ()⟩ - server.serve let owner ← TaskSet.new let res : OnceRef (Option (Except Error Unit)) ← OnceRef.new none owner.spawn do - res.set (some (← requestPeer client.newStream (← Limiter.new) m tc.clock 300 (some (noise 3 4)) + res.set (some (← requestPeer (openStreamOf p) (← Limiter.new) m tc.clock 300 (some (noise 3 4)) fun _ => pure ())) r.check "sim: a silent peer, the deadline arms" (← tc.nextTimer) tc.advance 300 owner.joinAll r.check "sim: a silent peer fails the request at its deadline" ((← res.get).any (failsWith .timeout)) tc.close - client.close - server.close - client.join - server.join + p.close -- The connection closes under a parked read: the request ends. do let tc ← TestClock.new - let (client, server) ← hostPair + let p ← hostPair let parked : Chan Unit ← Chan.new 1 - server.setHandler ⟨m.id, fun _ _ => ExceptT.mk do + p.server.setHandler ⟨m.id, fun _ _ => ExceptT.mk do discard <| parked.send () return .ok ()⟩ - server.serve let owner ← TaskSet.new let res : OnceRef (Option (Except Error Unit)) ← OnceRef.new none owner.spawn do - res.set (some (← requestPeer client.newStream (← Limiter.new) m tc.clock 60000 (some (noise 3 4)) + res.set (some (← requestPeer (openStreamOf p) (← Limiter.new) m tc.clock 60000 (some (noise 3 4)) fun _ => pure ())) -- The handler ran, so the request is past its write and parks on the read. discard <| parked.recv - client.close + p.client.close owner.joinAll r.check "sim: a connection closed under a parked read ends the request" ((← res.get).any fun e => match e with | .error _ => true | .ok _ => false) - server.close - client.join - server.join + p.server.close tc.close def run (r : Runner) : IO Unit := do diff --git a/packages/libp2p/host/Libp2p/Host.lean b/packages/libp2p/host/Libp2p/Host.lean index 569462a..c15d17f 100644 --- a/packages/libp2p/host/Libp2p/Host.lean +++ b/packages/libp2p/host/Libp2p/Host.lean @@ -5,6 +5,8 @@ import Libp2p.Host.Dedup import Libp2p.Host.Peerstore import Libp2p.Host.Policy import Libp2p.Host.Gate +import Libp2p.Host.Upgrade +import Libp2p.Host.StreamSelect import Libp2p.Host.Swarm import Libp2p.Host.Shutdown diff --git a/packages/libp2p/host/Libp2p/Host/StreamSelect.lean b/packages/libp2p/host/Libp2p/Host/StreamSelect.lean new file mode 100644 index 0000000..b2ad552 --- /dev/null +++ b/packages/libp2p/host/Libp2p/Host/StreamSelect.lean @@ -0,0 +1,82 @@ +import Streams +import Bytes +import Libp2p.Core +import Libp2p.Negotiate + +/-! +# Protocol selection on a muxed stream + +Every stream the muxer opens starts with its own multistream-select +1.0.0 exchange (the connection spec, "Opening new streams over a +multiplexed connection"): the initiator proposes the protocol ids in +order, and the responder answers with the first one in its handler +table. This module runs that exchange on a `Streams.Stream`. + +The negotiate package drives the same pure core over a +`Streams.Connection` (`Negotiate.negotiate`), for the security and muxer +stages of the upgrade. A muxed stream is a different seam: it reads and +writes through `Except` answers, and it carries a read-back buffer for +the leftover. The host owns this shell, and it reuses the core +unchanged: `initiatorStart`, `responderStart`, `drive`, and +`mapNegotiateError` all come from `Libp2p.Negotiate`. + +The bytes that follow the selection on the same read belong to the +protocol. `Stream.prependBytes` hands them on to the handler +(ARCHITECTURE §5.6). + +The read parks for as long as the peer stays silent, so no measure +bounds the loop (root §3.3). The `repeat` form keeps the definition free +of a `partial` marker, as in `Negotiate.negotiate`. +-/ + +set_option autoImplicit false + +namespace Libp2p.Host + +open Libp2p.Negotiate (initiatorStart responderStart drive mapNegotiateError) + +/-- Write each frame payload of the core to the stream, in order. A +failed write is the `closed` arm: the stream is gone. -/ +private def sendWrites (stream : Streams.Stream) (ws : List (List Nat)) : + ExceptT Libp2p.Core.Error Conc.Async Unit := do + for w in ws do + if let .error e ← stream.write (Bytes.ofNatList w) then + throw (.closed s!"stream negotiation write failed: {repr e}") + +/-- Negotiate a protocol on a muxed stream. The answer is the selected +id and the stream with the bytes after the selection put back in front +of its reads. -/ +def negotiateStream (role : Libp2p.Core.Role) (stream : Streams.Stream) + (ids : List Libp2p.Core.ProtocolId) : + ExceptT Libp2p.Core.Error Conc.Async (Libp2p.Core.ProtocolId × Streams.Stream) := do + let start := match role with + | .initiator => initiatorStart ids + | .responder => responderStart ids + let (initial, first) ← match start with + | .error e => throw (mapNegotiateError e) + | .ok pair => pure pair + sendWrites stream first + let mut state := initial + let mut input : List Nat := [] + repeat + match drive 1 state input with + | .error e => throw (mapNegotiateError e) + | .ok (.selected p _ rest writes) => + sendWrites stream writes + return (p, ← Streams.Stream.prependBytes stream (Bytes.ofNatList rest)) + | .ok (.running s _ rest writes) => + sendWrites stream writes + state := s + input := rest + | .ok (.waiting s _ rest writes) => + sendWrites stream writes + state := s + match ← stream.read with + | .error e => throw (.closed s!"stream negotiation read failed: {repr e}") + | .ok none => throw (.closed "stream ended during negotiation") + | .ok (some bytes) => input := rest ++ Bytes.toNatList bytes + -- Every arm above returns or loops, so the `repeat` never falls + -- through. The tail gives the `do` block its result type. + throw (.closed "stream negotiation ended without a selection") + +end Libp2p.Host diff --git a/packages/libp2p/host/Libp2p/Host/Swarm.lean b/packages/libp2p/host/Libp2p/Host/Swarm.lean index 9711349..a504524 100644 --- a/packages/libp2p/host/Libp2p/Host/Swarm.lean +++ b/packages/libp2p/host/Libp2p/Host/Swarm.lean @@ -3,8 +3,9 @@ import Conc import Streams import Libp2p.Identity import Libp2p.Core -import Libp2p.Negotiate.Legacy import Libp2p.Host.Error +import Libp2p.Host.StreamSelect +import Libp2p.Host.Upgrade import Libp2p.Host.Registry import Libp2p.Host.Dedup import Libp2p.Host.Peerstore @@ -36,7 +37,7 @@ ARCHITECTURE, "Refinements fixed at implementation (unit 1b)"): `allowDial` before a dial, `allowInbound` after an accept, the resource manager's `openConnection` before the upgrade, and `allowSecured` and `ConnectionScope.setPeer` inside the upgrade, after -the handshake and before the muxer stage (`Negotiate.upgradeWith`). A +the handshake and before the muxer stage (`Libp2p.Host.upgrade`). A refusal is `Error.gated` with the checkpoint. ## Ownership @@ -69,7 +70,7 @@ set_option autoImplicit false namespace Libp2p.Host -open Conc Streams Multiformats Libp2p.Identity Libp2p.Core Libp2p.Negotiate.Legacy +open Conc Streams Multiformats Libp2p.Identity Libp2p.Core /-- A failure of a task that no caller awaits: an inbound upgrade, a session that ended badly, an inbound stream. -/ @@ -242,9 +243,9 @@ private def admit (h : Host) (role : Role) (scope : ConnectionScope) (conn : Con /-! ## The upgrade -/ /-- Run the upgrade of `conn` as a pending connection: registered for -the shutdown (§5.5), under the deadline of the upgrader, with the -checkpoint. Every failure releases `scope`; the upgrader has already -closed the connection. -/ +the shutdown (§5.5), under the deadline of the upgrade, with the +checkpoint. Every failure releases `scope`. On a failure, the upgrade +already closed the connection. -/ private def upgradeConn (h : Host) (role : Role) (conn : Connection) (expected : Option PeerId) (scope : ConnectionScope) : Async (Except Error Upgraded) := do let key ← (h.draw : BaseIO Nat) @@ -257,7 +258,7 @@ private def upgradeConn (h : Host) (role : Role) (conn : Connection) (expected : conn.closeQuietly let unregister : BaseIO Unit := h.pending.update (·.map (·.erase key)) let r ← try - (upgradeAdmit h.config.clock role conn .empty expected (h.admit role scope conn) + (upgrade h.config.clock role conn expected (h.admit role scope conn) h.config.security h.config.muxers).run catch e => unregister diff --git a/packages/libp2p/host/Libp2p/Host/Upgrade.lean b/packages/libp2p/host/Libp2p/Host/Upgrade.lean new file mode 100644 index 0000000..31b7431 --- /dev/null +++ b/packages/libp2p/host/Libp2p/Host/Upgrade.lean @@ -0,0 +1,144 @@ +import Conc +import Streams +import Libp2p.Identity +import Libp2p.Core +import Libp2p.Negotiate +import Libp2p.Host.Error + +/-! +# The host's upgrade + +The upgrade of one connection, with the peer checkpoints in the place +the core seam fixes: `ConnectionGater.allowSecured` runs after the +handshake and before the muxer stage (`Libp2p/Core/Policy.lean`), as +go-libp2p's upgrader calls `InterceptSecured` +(`vendor/go-libp2p/p2p/net/upgrader/upgrader.go`, line 176). + +`Negotiate.upgradeWith` runs the same stages with no checkpoint. The +host composes the stages from the public functions of +`libp2p/negotiate` instead: `negotiate` for the security and the muxer +stages, `checkIdentity`, and `settleUpgrade`. The stage order follows +`Negotiate.upgradeCore`. A change to the order in one module needs the +same change in the other. + +The whole upgrade runs under one `Conc.withinDeadline` of +`Streams.unauthDeadlineMs` (root §5, invariant 4). A refusal at a +checkpoint is an answer, and no failure: the caller maps it to +`Host.Error.gated`. +-/ + +set_option autoImplicit false + +namespace Libp2p.Host + +open Conc Streams Libp2p.Identity Libp2p.Core + +/-- An upgraded connection: the muxer session and the peer the +handshake proved. The session's `join` also waits for the secure +connection under it, so `Upgraded.join` is the join of both. -/ +structure Upgraded where + /-- The muxer session. -/ + session : MuxSession + /-- The peer the handshake proved. -/ + remotePeer : PeerId + +/-- Wait for the session and the secure connection under it. -/ +def Upgraded.join (u : Upgraded) : ExceptT Core.Error Async Unit := + u.session.join + +/-- The answer of an upgrade that ran to its end: a session, or the +checkpoint that refused the proved peer. -/ +inductive Admission where + | upgraded (u : Upgraded) + | refused (at_ : Checkpoint) + +/-- Close a secure connection and wait for its tasks. The function +drops the answer of the join, because the caller already holds the +answer it reports. -/ +private def discardSecure (sc : SecureConn) : Async Unit := do + sc.toConnection.closeQuietly + discard sc.join.run + +/-- The session `s`, with a `join` that also waits for the secure +connection under it. The join waits for the secure connection even +when the session's join throws a fault, so none of its tasks outlives +the join (PRINCIPLES 3.2). The first failure wins. -/ +private def ownSecure (s : MuxSession) (sc : SecureConn) : MuxSession := + { s with + join := ExceptT.mk do + let a : Except IO.Error (Except Core.Error Unit) ← + try pure (.ok (← s.join.run)) catch e => pure (.error e) + let b ← sc.join.run + match a with + | .error e => throw e + | .ok a => return a.bind fun () => b } + +/-- The stages: negotiate the security protocol, run the handshake, +check the identity, ask `admit` about the proved peer, negotiate the +muxer, open the session. A failure or a refusal after the handshake +closes and joins the secure connection. -/ +private def stages (role : Role) (conn : Connection) (expected : Option PeerId) + (admit : PeerId → BaseIO (Option Checkpoint)) + (secs : List SecurityProtocol) (muxers : List Muxer) : + Async (Except Core.Error Admission) := do + match ← (Negotiate.negotiate role (BufReader.new conn) conn .empty (secs.map (·.id))).run with + | .error e => return .error e + | .ok stage => + let some sec := secs.find? (fun s => decide (s.id = stage.proto)) + | return .error (.protocol "selected security protocol is not configured") + match ← (sec.handshake role conn stage.leftover).run with + | .error e => return .error e + | .ok sc => + if let .error e := Negotiate.checkIdentity expected sc.remotePeer then + discardSecure sc + return .error e + -- The checkpoint: the handshake proved the peer, and no muxer + -- byte moves before this point. + if let some c ← admit sc.remotePeer then + discardSecure sc + return .ok (.refused c) + -- The secure `read` delivers the handshake's leftover, so the + -- muxer stage starts with no bytes. + let secure := sc.toConnection + match ← (Negotiate.negotiate role (BufReader.new secure) secure .empty + (muxers.map (·.id))).run with + | .error e => discardSecure sc; return .error e + | .ok stage2 => + let some mux := muxers.find? (fun m => decide (m.id = stage2.proto)) + | discardSecure sc + return .error (.protocol "selected muxer is not configured") + match ← (mux.«open» role secure stage2.leftover).run with + | .error e => discardSecure sc; return .error e + | .ok session => + return .ok (.upgraded { session := ownSecure session sc, remotePeer := sc.remotePeer }) + +/-- Close a session that finished after the timer won, and give its +join to `late`, the task set `upgrade` joins. No caller receives that +session. -/ +private def releaseLate (late : TaskSet) : Except Core.Error Admission → BaseIO Unit + | .ok (.upgraded u) => do + discard u.session.close.toBaseIO + late.spawn (discard u.join.run) + | _ => pure () + +/-- The upgrade of `conn` under the deadline, on `clock`. The expiry +closes the connection once, at the boundary. `Negotiate.settleUpgrade` +maps the expiry to the `timeout` arm. On a failure inside the stages, +this function closes the connection. -/ +def upgrade (clock : Clock) (role : Role) (conn : Connection) (expected : Option PeerId) + (admit : PeerId → BaseIO (Option Checkpoint)) + (secs : List SecurityProtocol) (muxers : List Muxer) : + ExceptT Core.Error Async Admission := do + let late ← TaskSet.new + let answer ← withinDeadline clock Streams.unauthDeadlineMs conn.closeQuietly + (stages role conn expected admit secs muxers) (releaseLate late) + late.joinAll + match Negotiate.settleUpgrade answer with + | .ok a => return a + | .error e => + -- An error from the boundary means the expiry hook closed the + -- connection. A failure inside the work leaves it open. + if answer matches .ok _ then conn.closeQuietly + throw e + +end Libp2p.Host diff --git a/packages/libp2p/host/docs/ARCHITECTURE.md b/packages/libp2p/host/docs/ARCHITECTURE.md index 10b82a9..a9eb11e 100644 --- a/packages/libp2p/host/docs/ARCHITECTURE.md +++ b/packages/libp2p/host/docs/ARCHITECTURE.md @@ -24,7 +24,8 @@ libp2p group is §7, row 10). | `Libp2p/Host/Swarm.lean` | `Config`, the `Host` state, `dial`, `dialPeer`, `startListening`, `newStream`, the accept loop, the upgrade pipeline, and stream serving | | `Libp2p/Host/Shutdown.lean` | `Host.close`, the §5.5 sequence | | `Libp2p/Host/Policy.lean` | the permissive default gater, and the permissive and denying test doubles the core pair assigned here; the default resource manager (admits everything) and the default ranker (keeps the order, no delay) | -| `Libp2p/Host/Mock/Negotiate.lean` | the mock of the W5.5 surface the host calls, until the host swap of issue #70 (see "The mock") | +| `Libp2p/Host/Upgrade.lean` | `upgrade` and `Upgraded`: the stages of `libp2p/negotiate`, with the peer checkpoint between the handshake and the muxer stage, under one deadline | +| `Libp2p/Host/StreamSelect.lean` | `negotiateStream`, multistream-select on a muxed stream, a shell over the pure core of `libp2p/negotiate` | All declarations live in the `Libp2p.Host` namespace, and the root module re-exports the public submodules. @@ -237,9 +238,16 @@ fixes them as follows. `allowInbound`, then `openConnection`, after an accept (`vendor/go-libp2p/p2p/net/upgrader/listener.go`, lines 183 and 193); and `allowSecured`, then `ConnectionScope.setPeer`, inside the - upgrade, through `Negotiate.upgradeWith` - (`vendor/go-libp2p/p2p/net/upgrader/upgrader.go`, line 176). A - refusal is `gated` with the checkpoint, and the scope is released. + upgrade, after the handshake and before the muxer stage + (`vendor/go-libp2p/p2p/net/upgrader/upgrader.go`, line 176, and the + core seam `ConnectionGater.allowSecured`). A refusal is `gated` with + the checkpoint, and the host releases the scope. +- **The host composes the upgrade.** `Negotiate.upgradeWith` takes no + checkpoint. `Libp2p/Host/Upgrade.lean` runs the stages from the + public functions of `libp2p/negotiate`: `negotiate`, `checkIdentity`, + and `settleUpgrade`. The stages keep the order of the upgrader, and + the checkpoint sits between the handshake and the muxer stage. One + `withinDeadline` bounds the whole upgrade, as in `upgradeWith`. The policy shapes are this unit's refinement of the core seams (the core ARCHITECTURE, "The policy shapes"). - **The answers of the dial path.** `dial` and `dialPeer` answer the @@ -265,9 +273,13 @@ fixes them as follows. - **Stream serving.** The serve task of a session accepts the peer's streams, and each stream runs in a task that the serve task owns and joins when the session ends. The stream negotiates against the - handler ids through `Negotiate.negotiateStream`, which puts the - bytes it read past the selecting message at the front of the stream - (§4.7). A failed negotiation, an id whose handler left the table, + handler ids through `negotiateStream`, which puts the bytes it read + past the selecting message at the front of the stream (§4.7). The + upgrader's `Negotiate.negotiate` runs over a `Connection`, and a + muxed stream is a `Stream`, so the host holds this shell + (`Libp2p/Host/StreamSelect.lean`). It drives the same pure core: + `initiatorStart`, `responderStart`, and `drive`. A failed + negotiation, an id whose handler left the table, and a handler failure reset the stream; the session stays. go-libp2p resets a stream whose negotiation fails in the same place (`vendor/go-libp2p/p2p/host/basic/basic_host.go`, line 347). A @@ -298,29 +310,13 @@ fixes them as follows. 1b milestone. A fault of a joined task does not stop the sequence, and `close` throws the first fault at the end. -## The mock - -The seam types come from `libp2p/core`. `libp2p/negotiate` takes the -real core too (issue #53), so its upgrader takes the values the host -holds. The host keeps a mock of the negotiate surface, -`Libp2p/Host/Mock/Negotiate.lean`, until the host swap of issue #70. -Its operations are `opaque`, and each body fails with the name of the -package; it states no axiom. Swarm opens `Libp2p.Host.Mock`, so -`Negotiate.upgradeWith` resolves as it will against the real package. -The swap (issue #70) is a retype: the merged upgrader takes no `admit` -checkpoint and exports no `negotiateStream`. - -Every upgrade runs through the negotiate mock and fails there, so the -unit 1b milestones run after the host swap of issue #70. The Noise and -yamux values are on the real core (issues #76 and #77). Unit 1a runs -in full. - ## What the package must not depend on - Inside libp2p, the chain puts everything below the host at its disposal (§2, §7 row 10). The host imports `identity` for the id and key types, `core` for the seam types, and `negotiate` for the - upgrader and the stream-negotiation responder its paths call. The + stages `upgrade` composes and the pure core that `negotiateStream` + drives. The security and mux implementations arrive inside the configuration lists (§3.1), so a host module that names a concrete seam value is a review failure. Test modules import diff --git a/packages/libp2p/host/docs/PLAN.md b/packages/libp2p/host/docs/PLAN.md index c058f57..17adc85 100644 --- a/packages/libp2p/host/docs/PLAN.md +++ b/packages/libp2p/host/docs/PLAN.md @@ -24,15 +24,14 @@ so unit 1a lands while the upgrader finishes. W5.2 sits outside the root row and reaches this unit through the root chain, which finishes it before W5.5. -Units 0 and 1a are done. Unit 1b's code builds over `libp2p/core` -and `libp2p/identity`, and against a mock of W5.5 (the package -ARCHITECTURE, "The mock"). The suite is `Tests/Host.lean`; the dedup -rule also carries theorems in `Libp2p/Host/Dedup.lean`. A host -upgrades through `libp2p/negotiate`, so the unit 1b milestones run at -the integration, and the suite names each one as a skip until then. Unit 1b needs two -calls from `libp2p/negotiate` beside `upgrade`: the post-handshake -checkpoint (`upgradeWith`) and the stream negotiation -(`negotiateStream`). The negotiate pair records both. +Units 0 and 1a are done. Unit 1b's code builds over `libp2p/core`, +`libp2p/identity`, and `libp2p/negotiate`. The suite is +`Tests/Host.lean`. The dedup rule also carries theorems in +`Libp2p/Host/Dedup.lean`. The host composes the upgrade from the +public stages of `libp2p/negotiate`, with the peer checkpoint after +the handshake (`Libp2p/Host/Upgrade.lean`). The stream negotiation is +the host's own shell, `negotiateStream`, over the pure core of +`libp2p/negotiate` (the package ARCHITECTURE, "Stream serving"). ## Consumers @@ -80,8 +79,8 @@ checkpoint (`upgradeWith`) and the stream negotiation ## Test milestones (unit 1b) - The upgrade pipeline over the sim, in both directions: dial and - listen through the real `Noise.securityProtocol` value, or the - fixed-key test value, and `Mux.yamux` (PRINCIPLES 1.4). An + listen through the fixed-key test value `Noise.fixedKeyProtocol` and + `Mux.yamux`, on a test clock (root §3.1, PRINCIPLES 1.4). An identity-check mismatch fails with the `Host.Error` arm that wraps `Core.Error.authentication` (§4.9; the negotiate pair). - The silent-peer regression at host level (PRINCIPLES 5.4): a raw diff --git a/packages/libp2p/negotiate/Libp2p/Negotiate/Legacy.lean b/packages/libp2p/negotiate/Libp2p/Negotiate/Legacy.lean deleted file mode 100644 index bc5f461..0000000 --- a/packages/libp2p/negotiate/Libp2p/Negotiate/Legacy.lean +++ /dev/null @@ -1,14 +0,0 @@ -import Libp2p.Negotiate.Legacy.Constants -import Libp2p.Negotiate.Legacy.Message -import Libp2p.Negotiate.Legacy.Step -import Libp2p.Negotiate.Legacy.Upgrade - -/-! -# The legacy negotiate surface - -The host of this branch still runs the pre-seam negotiate API: `upgradeWith` with the -checkpoint hook and `negotiateStream` per muxed stream. The reviewed tip carries the seam -versions of the negotiate package (issue #70 tracks the host swap), so this module keeps the -legacy entry points beside them, under their own namespace. Host files import this module -instead of `Libp2p.Negotiate`. --/ diff --git a/packages/libp2p/negotiate/Libp2p/Negotiate/Legacy/Constants.lean b/packages/libp2p/negotiate/Libp2p/Negotiate/Legacy/Constants.lean deleted file mode 100644 index 13e0452..0000000 --- a/packages/libp2p/negotiate/Libp2p/Negotiate/Legacy/Constants.lean +++ /dev/null @@ -1,22 +0,0 @@ -import Libp2p.Core - -/-! -# The constants of the negotiation - -One home per fact (root §3.8). The multistream protocol id and the -1 KiB message cap live only here. --/ - -set_option autoImplicit false - -namespace Libp2p.Negotiate.Legacy - -/-- The multistream 1.0.0 protocol id. The only protocol id literal of -the package (root §3.4). -/ -def protocolId : Libp2p.Core.ProtocolId := ⟨"/multistream/1.0.0"⟩ - -/-- The 1 KiB multistream cap (root §5.1, contract 1 of the package -docs). -/ -def maxMsg : Nat := 1024 - -end Libp2p.Negotiate.Legacy \ No newline at end of file diff --git a/packages/libp2p/negotiate/Libp2p/Negotiate/Legacy/Message.lean b/packages/libp2p/negotiate/Libp2p/Negotiate/Legacy/Message.lean deleted file mode 100644 index bb2d68c..0000000 --- a/packages/libp2p/negotiate/Libp2p/Negotiate/Legacy/Message.lean +++ /dev/null @@ -1,461 +0,0 @@ -import Libp2p.Negotiate.Legacy.Constants -import Varint -import Bytes - -/-! -# The multistream message - -`Msg` is one multistream message: a uvarint length prefix, then the -UTF-8 bytes of the text plus a trailing newline. The length counts the -newline (ARCHITECTURE §3.2, package docs "Public surface"). - -The decoder splits one frame at the index the prefix computes, then -runs the computed inverse of the payload map: it reads the body as -UTF-8, re-encodes it, and compares. `Bytes.toString?` rejects bad -UTF-8. The comparison rejects a missing trailing newline and a model -value at or above 256. So the `encodeDecode` proof needs no UTF-8 -reasoning. - -The 1 KiB cap is checked in `splitFrame`, before any payload -inspection (contract 1). The `ls` reply framing walks its entries as -ordinary `Msg` frames on a fuel counter, so the entry reader reuses -the proved frame decoder. --/ - -set_option autoImplicit false - -namespace Libp2p.Negotiate.Legacy - -open Bytes (DecodeError) - -/-! ## The payload map and its inverse -/ - -/-- The model bytes of a text, newline included. The one total map -between texts and payloads. -/ -def payloadOf (s : String) : List Nat := - Bytes.toNatList (Bytes.ofString (s ++ "\n")) - -/-- A total `UVarint` constructor for a length. `initiatorStart` and -`responderStart` keep the upgrade's lengths at or under `maxMsg`; the -modulo keeps the constructor total for any input. -/ -def boundedVarint (n : Nat) : Varint.UVarint := - ⟨n % Varint.limit, Nat.mod_lt _ (by decide)⟩ - -/-- The total text encoder: a uvarint length prefix, then the payload. -/ -def encodeText (s : String) : List Nat := - Varint.encode (boundedVarint (payloadOf s).length) ++ payloadOf s - -/-- A multistream message. `text` excludes the trailing newline. -`fits` is the proof-carrying cap fact of root §3.7: the message fits -the 1 KiB cap, so the encoder writes the exact length. -/ -structure Msg where - text : String - fits : (Bytes.ofString (text ++ "\n")).size ≤ maxMsg - deriving Repr, DecidableEq - -/-- The model payload length equals the byte size of the text with its -newline. -/ -theorem payloadOfLength (s : String) : - (payloadOf s).length = (Bytes.ofString (s ++ "\n")).size := by - simp only [payloadOf, Bytes.toNatList, List.length_map, Array.length_toList] - rfl - -/-- The payload of a text is its UTF-8 bytes plus the newline byte. -/ -theorem payloadOfEq (s : String) : - payloadOf s = Bytes.toNatList (Bytes.ofString s) ++ [10] := by - simp [payloadOf, Bytes.toNatList, Bytes.ofString, String.toByteArray_append] - exact ⟨10, by decide, rfl⟩ - -/-- Every payload ends in a newline, so it is never empty. -/ -theorem payloadOfNe (s : String) : payloadOf s ≠ [] := by - rw [payloadOfEq]; simp - -/-- The checked constructor. The tests build messages through it. -/ -def Msg.ofText? (s : String) : Option Msg := - if h : (Bytes.ofString (s ++ "\n")).size ≤ maxMsg then some ⟨s, h⟩ else none - -/-- A text whose frame fits the 1 KiB cap. `initiatorStart` and -`responderStart` call this before any write, so a configured id that -would produce an over-cap frame is a protocol error rather than a frame -the decoder refuses (root §5.1). -/ -def fitsText (s : String) : Bool := - decide ((Bytes.ofString (s ++ "\n")).size ≤ maxMsg) - -/-- The frame the codec writes for a message. -/ -def encodeMsg (m : Msg) : List Nat := encodeText m.text - -/-- The multistream header message. The literal lives in `Constants`. -/ -def header : Msg := ⟨protocolId.toString, by decide⟩ - -/-! ## The frame split -/ - -/-- Split one frame at the index the length prefix computes. The cap -is checked first, before any payload inspection (contract 1). A zero -declared length is malformed. A payload cut short is truncated, so -the caller keeps waiting. -/ -def splitFrame (bs : List Nat) : Except DecodeError (List Nat × List Nat) := - match Varint.decode? bs with - | .error e => .error e - | .ok (v, rest) => - if maxMsg < v.val then .error .malformed - else if v.val = 0 then .error .malformed - else if rest.length < v.val then .error .truncated - else .ok (rest.take v.val, rest.drop v.val) - -/-- The computed inverse of `payloadOf`, checked by re-encoding. -/ -def textOf? (payload : List Nat) : Option String := - match Bytes.toString? (Bytes.ofNatList payload.dropLast) with - | none => none - | some t => if payloadOf t = payload then some t else none - -/-- A text read back from a payload re-encodes to that payload. -/ -theorem textOf?Spec {payload : List Nat} {t : String} (h : textOf? payload = some t) : - payloadOf t = payload := by - unfold textOf? at h - split at h - · simp at h - · split at h - · simp_all - · simp at h - -/-- The inverse reads its own image back. -/ -theorem textOf?PayloadOf (s : String) : textOf? (payloadOf s) = some s := by - unfold textOf? - have hd : (payloadOf s).dropLast = Bytes.toNatList (Bytes.ofString s) := by - rw [payloadOfEq]; simp - rw [hd, Bytes.ofNatListToNatList, Bytes.toString?OfString] - simp - -/-- A payload whose last byte is not the newline reads back as no -text. -/ -theorem textOf?None {payload : List Nat} (hlast : payload.getLast? ≠ some 10) : - textOf? payload = none := by - unfold textOf? - cases hto : Bytes.toString? (Bytes.ofNatList payload.dropLast) with - | none => rfl - | some t => - by_cases heq : payloadOf t = payload - · have hpl : (payloadOf t).getLast? = some 10 := by rw [payloadOfEq]; simp - rw [heq] at hpl - exact (hlast hpl).elim - · simp [heq] - -/-- Decode one message and the bytes left after it. -/ -def decodeMsg? (bs : List Nat) : Except DecodeError (Msg × List Nat) := - match splitFrame bs with - | .error e => .error e - | .ok (payload, rest) => - match textOf? payload with - | none => .error .malformed - | some t => - if hf : (Bytes.ofString (t ++ "\n")).size ≤ maxMsg then .ok (⟨t, hf⟩, rest) - else .error .malformed - -/-! ## The frame lemmas -/ - -/-- One shared lemma: a framed payload splits back into that payload -and the remainder. -/ -theorem splitFrameEncode (s : String) (h : (payloadOf s).length ≤ maxMsg) (rest : List Nat) : - splitFrame (encodeText s ++ rest) = .ok (payloadOf s, rest) := by - have hlim : (payloadOf s).length < Varint.limit := by - unfold Varint.limit Varint.maxBytes maxMsg at *; omega - have hb : (boundedVarint (payloadOf s).length).val = (payloadOf s).length := by - simp [boundedVarint, Nat.mod_eq_of_lt hlim] - have hpos : 0 < (payloadOf s).length := List.length_pos_iff.mpr (payloadOfNe s) - unfold splitFrame encodeText - rw [List.append_assoc, Varint.decodeEncodeSuffix] - simp only [hb] - have h1 : ¬ maxMsg < (payloadOf s).length := by omega - have h2 : ¬ (payloadOf s).length = 0 := by omega - simp [h1, h2] - -/-- The suffix round trip: a message decodes back from its frame plus -any remainder. The lazy dialer's coalesced read needs this form. The -frame is `encodeText m.text`, definitionally `encodeMsg m`. -/ -theorem decodeMsgSuffix (m : Msg) (rest : List Nat) : - decodeMsg? (encodeText m.text ++ rest) = .ok (m, rest) := by - have h : (payloadOf m.text).length ≤ maxMsg := by rw [payloadOfLength]; exact m.fits - unfold decodeMsg? - rw [splitFrameEncode m.text h] - simp [textOf?PayloadOf, m.fits] - -/-- The inverse direction: every accepted input is the encoding of -the decoded value. No UTF-8 reasoning; `textOf?` already re-encoded. -/ -theorem decodeMsgEncodeDecode (bs : List Nat) (m : Msg) (rest : List Nat) - (h : decodeMsg? bs = .ok (m, rest)) : encodeMsg m ++ rest = bs := by - show encodeText m.text ++ rest = bs - unfold decodeMsg? splitFrame at h - split at h - · simp at h - · rename_i payload rest0 hsplit - split at hsplit - · simp at hsplit - · rename_i v r0 hdec - have hbs := Varint.encodeDecode hdec - split at hsplit - · simp at hsplit - split at hsplit - · simp at hsplit - split at hsplit - · simp at hsplit - rename_i hcap _ hlen - simp only [Except.ok.injEq, Prod.mk.injEq] at hsplit - obtain ⟨rfl, rfl⟩ := hsplit - split at h - · simp at h - · rename_i t ht - have hp := textOf?Spec ht - split at h - · simp only [Except.ok.injEq, Prod.mk.injEq] at h - obtain ⟨rfl, rfl⟩ := h - have hl : (payloadOf t).length = v.val := by - rw [hp]; simp; omega - have hv : boundedVarint (payloadOf t).length = v := by - cases v; simp_all [boundedVarint, Nat.mod_eq_of_lt] - rw [encodeText, hv, hp, List.append_assoc, List.take_append_drop] - exact hbs.symm - · simp at h - -/-- The codec writes the frame and reads one message back. -/ -instance : Bytes.Codec Msg where - encode := encodeMsg - decode? := decodeMsg? - -theorem decodeMsgEncode (m : Msg) : decodeMsg? (encodeMsg m) = .ok (m, []) := by - simpa [encodeMsg] using decodeMsgSuffix m [] - -/-- The round-trip law that §3.2 requires. -/ -instance : Bytes.LawfulCodec Msg where - decodeEncode := decodeMsgEncode - encodeDecode := by - intro bs a rest h - exact decodeMsgEncodeDecode bs a rest h - -/-- The cap fires before the payload is read, for every payload that -follows the prefix (contract 1). -/ -theorem decodeMsgCap {v : Varint.UVarint} {payload : List Nat} - (h : maxMsg < v.val) : - decodeMsg? (Varint.encode v ++ payload) = .error .malformed := by - simp [decodeMsg?, splitFrame, Varint.decodeEncodeSuffix, h] - -/-- A payload cut short is incomplete: more bytes may complete it. -/ -theorem decodeMsgTruncated {v : Varint.UVarint} {payload : List Nat} - (hv0 : 0 < v.val) (hcap : v.val ≤ maxMsg) (hshort : payload.length < v.val) : - decodeMsg? (Varint.encode v ++ payload) = .error .truncated := by - unfold decodeMsg? splitFrame - rw [Varint.decodeEncodeSuffix] - have h1 : ¬ maxMsg < v.val := by omega - have h2 : ¬ v.val = 0 := by omega - simp only [h1, h2, hshort, ite_true, ite_false] - -/-- A zero declared length is malformed. -/ -theorem decodeMsgZeroLength {v : Varint.UVarint} {payload : List Nat} - (h : v.val = 0) : - decodeMsg? (Varint.encode v ++ payload) = .error .malformed := by - simp [decodeMsg?, splitFrame, Varint.decodeEncodeSuffix, h, maxMsg] - -/-- A body whose last byte is not the newline is malformed. -/ -theorem decodeMsgNoNewline {v : Varint.UVarint} {payload : List Nat} - (hv0 : 0 < v.val) (hcap : v.val ≤ maxMsg) (hlen : payload.length = v.val) - (hlast : payload.getLast? ≠ some 10) : - decodeMsg? (Varint.encode v ++ payload) = .error .malformed := by - have hnone : textOf? (List.take v.val payload) = none := by - rw [← hlen, List.take_length] - exact textOf?None hlast - unfold decodeMsg? splitFrame - rw [Varint.decodeEncodeSuffix] - have h1 : ¬ maxMsg < v.val := by omega - have h2 : ¬ v.val = 0 := by omega - have h3 : ¬ payload.length < v.val := by omega - simp only [h1, h2, h3, ite_false] - rw [hnone] - -/-- The framing budget: a message frame spans its payload plus at -most nine prefix bytes. -/ -theorem encodeMsgLength (m : Msg) : - (encodeMsg m).length ≤ maxMsg + Varint.maxBytes := by - have h1 := Varint.lengthEncodeLe (boundedVarint (payloadOf m.text).length) - have h2 : (payloadOf m.text).length ≤ maxMsg := by rw [payloadOfLength]; exact m.fits - simp only [encodeMsg, encodeText, List.length_append] - omega - -/-! ## The `ls` reply framing -/ - -/-- One `ls` entry is an ordinary message frame: the entry's text is -the protocol id, and the frame carries the id plus its newline. -/ -def lsEntries (ids : List Libp2p.Core.ProtocolId) : List Nat := - (ids.map (fun p => encodeText p.toString)).flatten - -/-- The body of an `ls` reply: one entry frame per id, then the final -newline byte that ends the list. -/ -def lsBody (ids : List Libp2p.Core.ProtocolId) : List Nat := lsEntries ids ++ [10] - -/-- The full `ls` reply: the body wrapped in one uvarint length -prefix, exactly like a message. -/ -def encodeLs (ids : List Libp2p.Core.ProtocolId) : List Nat := - Varint.encode (boundedVarint (lsBody ids).length) ++ lsBody ids - -/-- Walk the entries of one complete `ls` body. The fuel counts -entries; the caller passes the body's byte length, which is at least -the entry count because every entry carries its newline. `[10]` alone -is the terminator, not a leading byte: a leading `0x0A` can be a -legitimate one-byte prefix for a nine-byte id. -/ -def decodeEntries : Nat → List Nat → Except DecodeError (List Libp2p.Core.ProtocolId) - | 0, _ => .error .malformed - | fuel + 1, bs => - if bs = [10] then .ok [] - else match decodeMsg? bs with - | .error _ => .error .malformed - | .ok (m, rest) => (decodeEntries fuel rest).map (⟨m.text⟩ :: ·) - -/-- Decode one `ls` reply frame. The outer prefix goes through the -same split and cap as a message. An entry that ends mid-body is -malformed, because only the outer prefix and length can be completed -by more bytes. -/ -def decodeLs? (bs : List Nat) : - Except DecodeError (List Libp2p.Core.ProtocolId × List Nat) := - match splitFrame bs with - | .error e => .error e - | .ok (body, rest) => (decodeEntries body.length body).map (·, rest) - -/-- An entry frame is at least as long as its payload. -/ -theorem encodeTextLength (s : String) : (payloadOf s).length ≤ (encodeText s).length := by - simp [encodeText] - -/-- The entries of an encoded list read back in order, under fuel and -cap bounds. -/ -theorem decodeEntriesLs (ids : List Libp2p.Core.ProtocolId) (fuel : Nat) - (hf : ids.length < fuel) (hb : (lsEntries ids).length ≤ maxMsg) : - decodeEntries fuel (lsEntries ids ++ [10]) = .ok ids := by - induction ids generalizing fuel with - | nil => - cases fuel with - | zero => simp at hf - | succ f => simp [decodeEntries, lsEntries] - | cons p ps ih => - cases fuel with - | zero => simp at hf - | succ f => - have hlen : (lsEntries (p :: ps)).length = - (encodeText p.toString).length + (lsEntries ps).length := by - simp [lsEntries] - have hfit : (Bytes.ofString (p.toString ++ "\n")).size ≤ maxMsg := by - rw [← payloadOfLength]; have := encodeTextLength p.toString; omega - have hne : lsEntries (p :: ps) ++ [10] ≠ [10] := by - intro h - have hl := congrArg List.length h - rw [List.length_append, hlen] at hl - have h0 := List.length_pos_iff.mpr (payloadOfNe p.toString) - have h1 := encodeTextLength p.toString - simp only [List.length_singleton] at hl - omega - have hsplit : lsEntries (p :: ps) ++ [10] = - encodeText (Msg.text ⟨p.toString, hfit⟩) ++ (lsEntries ps ++ [10]) := by - simp [lsEntries] - simp only [decodeEntries, hne, ite_false] - rw [hsplit, decodeMsgSuffix ⟨p.toString, hfit⟩ (lsEntries ps ++ [10])] - change (Except.map (fun x => (⟨p.toString⟩ : Libp2p.Core.ProtocolId) :: x) - (decodeEntries f (lsEntries ps ++ [10])) = .ok (p :: ps)) - rw [ih f (by simp at hf; omega) (by omega)] - rfl - -/-- The byte length of the entries is at least the id count. -/ -theorem lsEntriesLength (ids : List Libp2p.Core.ProtocolId) : - ids.length ≤ (lsEntries ids).length := by - induction ids with - | nil => simp [lsEntries] - | cons p ps ih => - have h0 := List.length_pos_iff.mpr (payloadOfNe p.toString) - have := encodeTextLength p.toString - simp [lsEntries] at ih ⊢; omega - -/-- The `ls` reply round trip. The length prefix alone delimits the -entries, so no cleanliness hypothesis is needed. -/ -theorem decodeLsEncodeLs (ids : List Libp2p.Core.ProtocolId) - (hbody : (lsBody ids).length ≤ maxMsg) : - decodeLs? (encodeLs ids) = .ok (ids, []) := by - have hlim : (lsBody ids).length < Varint.limit := by - unfold Varint.limit Varint.maxBytes maxMsg at *; omega - have hb : (boundedVarint (lsBody ids).length).val = (lsBody ids).length := by - simp [boundedVarint, Nat.mod_eq_of_lt hlim] - have hl := lsEntriesLength ids - unfold decodeLs? encodeLs splitFrame - rw [← List.append_nil (lsBody ids), Varint.decodeEncodeSuffix] - simp only [hb, List.append_nil] - have h1 : ¬ maxMsg < (lsBody ids).length := by omega - have h2 : ¬ (lsBody ids).length = 0 := by simp [lsBody] - simp only [h1, h2, ite_false, Nat.lt_irrefl, List.take_length, List.drop_length] - have hb2 : (lsEntries ids).length ≤ maxMsg := by simp [lsBody] at hbody; omega - rw [lsBody, decodeEntriesLs ids _ (by simp; omega) hb2] - rfl - -/-- A frame that splits back is exactly the prefix plus body plus -remainder. -/ -theorem splitFrameInv {bs body rest : List Nat} (h : splitFrame bs = .ok (body, rest)) : - Varint.encode (boundedVarint body.length) ++ body ++ rest = bs := by - unfold splitFrame at h - split at h - · simp at h - · rename_i v r0 hdec - have hbs := Varint.encodeDecode hdec - split at h - · simp at h - split at h - · simp at h - split at h - · simp at h - rename_i hcap _ hlen - simp only [Except.ok.injEq, Prod.mk.injEq] at h - obtain ⟨rfl, rfl⟩ := h - have hl : (r0.take v.val).length = v.val := by simp; omega - have hv : boundedVarint (r0.take v.val).length = v := by - cases v; simp_all [boundedVarint, Nat.mod_eq_of_lt] - rw [hv, List.append_assoc, List.take_append_drop] - exact hbs.symm - -/-- Entries that read back are exactly the encoding of the list. -/ -theorem decodeEntriesInv (f : Nat) (bs : List Nat) (ids : List Libp2p.Core.ProtocolId) - (h : decodeEntries f bs = .ok ids) : lsEntries ids ++ [10] = bs := by - induction f generalizing bs ids with - | zero => simp [decodeEntries] at h - | succ f ih => - unfold decodeEntries at h - split at h - · rename_i hb - simp only [Except.ok.injEq] at h - subst h - simp [lsEntries, hb] - · split at h - · simp at h - · rename_i m rest hm - cases hr : decodeEntries f rest with - | error e => simp [hr, Except.map] at h - | ok ids' => - simp [hr, Except.map] at h - subst h - have h1 := decodeMsgEncodeDecode bs m rest hm - have h2 := ih rest ids' hr - simp only [lsEntries, List.map_cons, List.flatten_cons, List.append_assoc] at h2 ⊢ - rw [h2] - exact h1 - -/-- The all-input inverse of the `ls` reply: every accepted input is -the encoding of the decoded list. -/ -theorem decodeLsInv {bs : List Nat} {ids : List Libp2p.Core.ProtocolId} {rest : List Nat} - (h : decodeLs? bs = .ok (ids, rest)) : encodeLs ids ++ rest = bs := by - unfold decodeLs? at h - split at h - · simp at h - · rename_i body r hs - cases he : decodeEntries body.length body with - | error e => simp [he, Except.map] at h - | ok ids' => - simp [he, Except.map] at h - obtain ⟨rfl, rfl⟩ := h - have hb := decodeEntriesInv _ _ _ he - have hs' := splitFrameInv hs - unfold encodeLs lsBody - rw [hb] - exact hs' - -end Libp2p.Negotiate.Legacy \ No newline at end of file diff --git a/packages/libp2p/negotiate/Libp2p/Negotiate/Legacy/Step.lean b/packages/libp2p/negotiate/Libp2p/Negotiate/Legacy/Step.lean deleted file mode 100644 index 8fedc1b..0000000 --- a/packages/libp2p/negotiate/Libp2p/Negotiate/Legacy/Step.lean +++ /dev/null @@ -1,509 +0,0 @@ -import Libp2p.Negotiate.Legacy.Message -import Libp2p.Core -import Bytes - -/-! -# The pure negotiation core - -`State`, `Event`, `Action`, and `step` are the pure state machine of -multistream-select (root §3.3). One `Event.bytes` feeds raw bytes, and -the core decodes exactly one frame per call, so the -malformed-versus-incomplete call stays in the core (§3.1). A decoded -frame goes through a private `classify` step that names the answer, -then a private `act` step that performs it. The theorems about the -chosen id reason over `classify`; the theorems about writes and phase -reason over `act`. - -The step returns `Except Error (State × List Action)`, the root §3.3 -shape. `Action` has no error constructor (package docs, "Decisions"): -the `Except` channel carries the failure (root §3.5). --/ - -set_option autoImplicit false - -namespace Libp2p.Negotiate.Legacy - -open Bytes (DecodeError) - -/-- The phase of one negotiation. -/ -inductive Phase where - | headerPending - | negotiating - | awaitingAnswer - | selected - deriving Repr, DecidableEq - -/-- The core state. `role` is the side. `supported` is the responder's -ordered list. `proposals` is the initiator's remaining list, in order. -`buffer` holds the bytes no frame has consumed yet. -/ -structure State where - role : Libp2p.Core.Role - supported : List Libp2p.Core.ProtocolId - proposals : List Libp2p.Core.ProtocolId - buffer : List Nat - phase : Phase - deriving Repr, DecidableEq - -/-- One input to the core. Raw bytes arrive as one event; a proposal -is the local dialer's input. -/ -inductive Event where - | bytes (bs : List Nat) - | propose (p : Libp2p.Core.ProtocolId) - deriving Repr, DecidableEq - -/-- One output of the core. A write carries framed bytes. A select -ends the negotiation with the chosen id and the leftover bytes. -/ -inductive Action where - | write (bs : List Nat) - | select (proto : Libp2p.Core.ProtocolId) (leftover : List Nat) - deriving Repr, DecidableEq - -/-- The replies the responder sends, the closed set of §3.4. -/ -inductive Reply where - | na - | ls - deriving DecidableEq, Repr, Hashable - -/-- The wire text of a reply. -/ -def Reply.text : Reply → String - | .na => "na" - | .ls => "ls" - -/-- The table the inverse searches. -/ -def Reply.all : List Reply := [.na, .ls] - -/-- The computed inverse of `Reply.text`: a text names a reply when it -equals the reply's text. -/ -def Reply.ofText? (s : String) : Option Reply := Reply.all.find? (fun r => r.text = s) - -/-- The table conformance: the inverse reads exactly the reply whose -text it is given. -/ -theorem Reply.ofText?EqSome {s : String} {r : Reply} : - Reply.ofText? s = some r ↔ r.text = s := by - constructor - · intro h - have := List.find?_some h - simpa using this - · intro h - subst h - have hmem : r ∈ Reply.all := by cases r <;> simp [Reply.all] - unfold Reply.ofText? - cases r <;> simp [Reply.all, Reply.text] - -/-- The reply as a message. The text is the message text. -/ -def Reply.toMsg : Reply → Msg - | .na => ⟨"na", by decide⟩ - | .ls => ⟨"ls", by decide⟩ - -/-- The message form of the reply recognition. -/ -def Reply.ofMsg? (m : Msg) : Option Reply := Reply.ofText? m.text - -/-- Why a negotiation failed. `decode` carries a frame the decoder -rejects. `negotiationFailed` covers a bad exchange, an exhausted -proposal list, and an invalid local list at start. -/ -inductive Error where - | decode (e : DecodeError) - | negotiationFailed - deriving Repr, DecidableEq, Hashable - -/-- The pure negotiation core of the spike: the first of `ours` that -`theirs` also supports. -/ -def choose (ours theirs : List Libp2p.Core.ProtocolId) : Option Libp2p.Core.ProtocolId := - List.find? theirs.contains ours - -/-- A chosen id is in our list and supported by theirs. -/ -theorem chooseSpec {ours theirs : List Libp2p.Core.ProtocolId} {p : Libp2p.Core.ProtocolId} - (h : choose ours theirs = some p) : p ∈ ours ∧ theirs.contains p = true := by - unfold choose at h - exact ⟨List.mem_of_find?_eq_some h, List.find?_some h⟩ - -/-! ## The answers -/ - -/-- What a decoded text means for this state. Byte-free, so the -theorems about the chosen id reason over this classification. -/ -private inductive Answer where - | echo - | listIds - | pick (p : Libp2p.Core.ProtocolId) - | refuse - | advance - | retry (q : Libp2p.Core.ProtocolId) (ps : List Libp2p.Core.ProtocolId) - | accept (p : Libp2p.Core.ProtocolId) - | fail - -/-- Classify a decoded text for a state. -/ -private def classify (s : State) (text : String) : Answer := - match s.role, s.phase with - | .responder, .headerPending => if text = protocolId.toString then .echo else .fail - | .responder, .negotiating => - match Reply.ofText? text with - | some .ls => .listIds - | _ => - if text = protocolId.toString then .advance - else if (⟨text⟩ : Libp2p.Core.ProtocolId) ∈ s.supported then .pick ⟨text⟩ - else .refuse - | .initiator, .negotiating => if text = protocolId.toString then .advance else .fail - | .initiator, .awaitingAnswer => - match s.proposals with - | [] => .fail - | p :: ps => - if Reply.ofText? text = some .na then - match ps with - | [] => .fail - | q :: qs => .retry q (q :: qs) - else if text = p.toString then .accept p else .fail - | _, _ => .fail - -/-- Perform an answer. It decodes nothing, so its theorems need no -decoder facts. -/ -private def act (s : State) (rest : List Nat) : Answer → Except Error (State × List Action) - | .echo => .ok ({ s with buffer := rest, phase := .negotiating }, [.write (encodeText protocolId.toString)]) - | .listIds => .ok ({ s with buffer := rest }, [.write (encodeLs s.supported)]) - | .pick p => .ok ({ s with buffer := [], phase := .selected }, - [.write (encodeText p.toString), .select p rest]) - | .refuse => .ok ({ s with buffer := rest }, [.write (encodeMsg (Reply.toMsg .na))]) - | .advance => - match s.role with - | .initiator => .ok ({ s with buffer := rest, phase := .awaitingAnswer }, []) - | .responder => .ok ({ s with buffer := rest }, []) - | .retry q ps => .ok ({ s with buffer := rest, proposals := ps }, [.write (encodeText q.toString)]) - | .accept p => .ok ({ s with buffer := [], phase := .selected }, [.select p rest]) - | .fail => .error .negotiationFailed - -/-- A proposal event. An initiator in the header phase with an empty list -records the id and writes the coalesced header plus the first proposal frame. -Any other proposal event leaves the state unchanged: `initiatorStart` seeds -the ordered list before the first frame moves, and the `na` retry walks that -list from `classify`. -/ -private def propose (s : State) (p : Libp2p.Core.ProtocolId) : Except Error (State × List Action) := - if s.role = .initiator then - if s.phase = .headerPending then - if s.proposals = [] then - .ok ({ s with proposals := [p], phase := .negotiating }, - [.write (encodeMsg header ++ encodeText p.toString)]) - else .ok (s, []) - else .ok (s, []) - else .ok (s, []) - -/-- One transition of the core. A `bytes` event appends to the buffer -and decodes one frame. `truncated` keeps the buffer and waits; -`malformed` is a protocol error; a decoded frame goes through -`classify` and `act`. -/ -def step (s : State) : Event → Except Error (State × List Action) - | .propose p => propose s p - | .bytes bs => - match decodeMsg? (s.buffer ++ bs) with - | .error .truncated => .ok ({ s with buffer := s.buffer ++ bs }, []) - | .error .malformed => .error (.decode .malformed) - | .ok (m, rest) => act s rest (classify s m.text) - -/-! ## The answer lemmas -/ - -/-- A step that selects came from the pick or accept answer. -/ -private theorem actSelect {s : State} {rest : List Nat} {v : Answer} {s' : State} - {acts : List Action} {p : Libp2p.Core.ProtocolId} {r : List Nat} - (h : act s rest v = .ok (s', acts)) (hm : Action.select p r ∈ acts) : - v = .pick p ∨ v = .accept p := by - cases v <;> simp only [act] at h - <;> ((try split at h) - <;> (try simp only [Except.ok.injEq, Prod.mk.injEq] at h) - <;> (try obtain ⟨-, rfl⟩ := h) - <;> (try simp_all)) - -/-- A pick comes from the responder's supported list. -/ -private theorem classifyPick {s : State} {t : String} {p : Libp2p.Core.ProtocolId} - (h : classify s t = .pick p) : p ∈ s.supported := by - unfold classify at h - repeat' split at h - all_goals simp_all - -/-- An accept comes from the head of the initiator's proposals. -/ -private theorem classifyAcceptInitiator {s : State} {t : String} {p : Libp2p.Core.ProtocolId} - (h : classify s t = .accept p) : p ∈ s.proposals := by - unfold classify at h - repeat' split at h - all_goals simp_all - -/-- An accept carries the accepted text. -/ -private theorem classifyAcceptText {s : State} {t : String} {p : Libp2p.Core.ProtocolId} - (h : classify s t = .accept p) : p.toString = t := by - unfold classify at h - repeat' split at h - all_goals simp_all - -/-- A responder never accepts. -/ -private theorem classifyAcceptResponder {s : State} {t : String} {p : Libp2p.Core.ProtocolId} - (hr : s.role = .responder) (h : classify s t = .accept p) : False := by - unfold classify at h - repeat' split at h - all_goals simp_all - -/-- An initiator never picks. -/ -private theorem classifyPickInitiator {s : State} {t : String} {p : Libp2p.Core.ProtocolId} - (hr : s.role = .initiator) (h : classify s t = .pick p) : False := by - unfold classify at h - repeat' split at h - all_goals simp_all - -/-- An initiator never sees the echo answer. -/ -private theorem classifyEchoInitiator {s : State} {t : String} - (hr : s.role = .initiator) (h : classify s t = .echo) : False := by - unfold classify at h - repeat' split at h - all_goals simp_all - -/-- An initiator never sees the list answer. -/ -private theorem classifyListIdsInitiator {s : State} {t : String} - (hr : s.role = .initiator) (h : classify s t = .listIds) : False := by - unfold classify at h - repeat' split at h - all_goals simp_all - -/-- An initiator never sees the refuse answer. -/ -private theorem classifyRefuseInitiator {s : State} {t : String} - (hr : s.role = .initiator) (h : classify s t = .refuse) : False := by - unfold classify at h - repeat' split at h - all_goals simp_all - -/-- The `ls` text in the negotiation draws the list answer. -/ -private theorem classifyListIds {s : State} {t : String} - (hrole : s.role = .responder) (hphase : s.phase = .negotiating) - (hls : Reply.ofText? t = some .ls) : classify s t = .listIds := by - unfold classify - rw [hrole, hphase] - rw [hls] - -/-- An unknown id in the negotiation draws the refuse answer. -/ -private theorem classifyRefuse {s : State} {t : String} - (hrole : s.role = .responder) (hphase : s.phase = .negotiating) - (hnot : (⟨t⟩ : Libp2p.Core.ProtocolId) ∉ s.supported) - (hheader : t ≠ protocolId.toString) (hreply : Reply.ofText? t = none) : - classify s t = .refuse := by - unfold classify - rw [hrole, hphase] - simp only [hreply] - rw [ite_eq_right hheader, ite_eq_right hnot] - -/-- A retry pops one proposal: the new list is the tail of the old. -/ -private theorem classifyRetryTail {s : State} {t : String} {q : Libp2p.Core.ProtocolId} - {ps : List Libp2p.Core.ProtocolId} (h : classify s t = .retry q ps) : - s.proposals.tail = ps := by - unfold classify at h - repeat' split at h - all_goals (cases h <;> simp_all [List.tail_cons]) - -/-- A write came from one of the five writing answers. -/ -private theorem actWriteAnswer {s : State} {rest : List Nat} {v : Answer} {s' : State} - {actions : List Action} {w : List Nat} - (h : act s rest v = .ok (s', actions)) (hw : Action.write w ∈ actions) : - v = .echo ∨ v = .listIds ∨ (∃ p, v = .pick p) ∨ v = .refuse ∨ - (∃ q ps, v = .retry q ps) := by - cases v with - | echo => exact Or.inl rfl - | listIds => exact Or.inr (Or.inl rfl) - | pick p => exact Or.inr (Or.inr (Or.inl ⟨p, rfl⟩)) - | refuse => exact Or.inr (Or.inr (Or.inr (Or.inl rfl))) - | retry q ps => exact Or.inr (Or.inr (Or.inr (Or.inr ⟨q, ps, rfl⟩))) - | advance => - simp only [act] at h - split at h <;> simp only [Except.ok.injEq, Prod.mk.injEq] at h - all_goals (obtain ⟨-, rfl⟩ := h; simp at hw) - | accept p => - simp only [act] at h - simp only [Except.ok.injEq, Prod.mk.injEq] at h - obtain ⟨-, rfl⟩ := h - simp at hw - | fail => - simp only [act] at h - cases h - -/-- The list answer writes the `ls` reply and stays in the phase. -/ -private theorem actListIds {s : State} {rest : List Nat} {s' : State} {actions : List Action} - (h : act s rest .listIds = .ok (s', actions)) : - actions = [.write (encodeLs s.supported)] ∧ s'.phase = s.phase ∧ s'.buffer = rest := by - unfold act at h - simp only [Except.ok.injEq, Prod.mk.injEq] at h - obtain ⟨hstate, hact⟩ := h - subst hact - rw [← hstate] - simp - -/-- The refuse answer writes the `na` frame and stays in the phase. -/ -private theorem actRefuse {s : State} {rest : List Nat} {s' : State} {actions : List Action} - (h : act s rest .refuse = .ok (s', actions)) : - actions = [.write (encodeMsg (Reply.toMsg .na))] ∧ s'.phase = s.phase ∧ s'.buffer = rest := by - unfold act at h - simp only [Except.ok.injEq, Prod.mk.injEq] at h - obtain ⟨hstate, hact⟩ := h - subst hact - rw [← hstate] - simp - -/-- The retry answer replaces the proposals with the remaining list. -/ -private theorem actRetryWrites {s : State} {rest : List Nat} {q : Libp2p.Core.ProtocolId} - {ps : List Libp2p.Core.ProtocolId} {s' : State} {actions : List Action} - (h : act s rest (.retry q ps) = .ok (s', actions)) : s'.proposals = ps := by - unfold act at h - simp only [Except.ok.injEq, Prod.mk.injEq] at h - obtain ⟨hstate, _⟩ := h - rw [← hstate] - -/-! ## The step theorems -/ - -/-- A responder selects only from its supported list (contract 2). -/ -theorem stepSelectResponderMem {s : State} {bs : List Nat} {s' : State} - {actions : List Action} {p : Libp2p.Core.ProtocolId} {rest : List Nat} - (hrole : s.role = .responder) - (hstep : step s (.bytes bs) = .ok (s', actions)) - (hmem : Action.select p rest ∈ actions) : p ∈ s.supported := by - simp only [step] at hstep - split at hstep - · simp only [Except.ok.injEq, Prod.mk.injEq] at hstep - obtain ⟨-, rfl⟩ := hstep - simp at hmem - · simp at hstep - · rcases actSelect hstep hmem with hv | hv - · exact classifyPick hv - · exact (classifyAcceptResponder hrole hv).elim - -/-- An initiator selects only from its own proposal list. -/ -theorem stepSelectInitiatorMem {s : State} {bs : List Nat} {s' : State} - {actions : List Action} {p : Libp2p.Core.ProtocolId} {rest : List Nat} - (hrole : s.role = .initiator) - (hstep : step s (.bytes bs) = .ok (s', actions)) - (hmem : Action.select p rest ∈ actions) : p ∈ s.proposals := by - simp only [step] at hstep - split at hstep - · simp only [Except.ok.injEq, Prod.mk.injEq] at hstep - obtain ⟨-, rfl⟩ := hstep - simp at hmem - · simp at hstep - · rcases actSelect hstep hmem with hv | hv - · exact (classifyPickInitiator hrole hv).elim - · exact classifyAcceptInitiator hv - -/-- A step that writes advances the initiator's proposal list by one -head: the new list is the old tail, so it is strictly shorter -(contract 2, the `na` retry). -/ -theorem stepInitiatorNaProgress {s : State} {bs : List Nat} {s' : State} - {actions : List Action} {p : Libp2p.Core.ProtocolId} {ps : List Libp2p.Core.ProtocolId} - {w : List Nat} - (hrole : s.role = .initiator) (hprop : s.proposals = p :: ps) - (hstep : step s (.bytes bs) = .ok (s', actions)) - (hwrite : Action.write w ∈ actions) : - s'.proposals = ps ∧ s'.proposals.length < s.proposals.length := by - simp only [step] at hstep - split at hstep - · simp only [Except.ok.injEq, Prod.mk.injEq] at hstep - obtain ⟨-, rfl⟩ := hstep - simp at hwrite - · simp at hstep - · rcases actWriteAnswer hstep hwrite with hc | hc | hc | hc | hc - · exact (classifyEchoInitiator hrole hc).elim - · exact (classifyListIdsInitiator hrole hc).elim - · rcases hc with ⟨p', hc⟩ - exact (classifyPickInitiator hrole hc).elim - · exact (classifyRefuseInitiator hrole hc).elim - · rcases hc with ⟨q, ps', hc⟩ - rw [hc] at hstep - have hnew : s'.proposals = ps' := actRetryWrites hstep - have htail : s.proposals.tail = ps' := classifyRetryTail hc - have hps : ps' = ps := by - rw [← htail, hprop] - rfl - constructor - · rw [hnew, hps] - · rw [hnew, hps, hprop] - simp only [List.length_cons] - omega - -/-- The `ls` answer writes the reply and stays in the negotiation -(contract 3). -/ -theorem stepResponderLsStays {s : State} {bs : List Nat} {s' : State} - {actions : List Action} {m : Msg} {rest : List Nat} - (hrole : s.role = .responder) (hphase : s.phase = .negotiating) - (hdec : decodeMsg? (s.buffer ++ bs) = .ok (m, rest)) - (hls : Reply.ofMsg? m = some .ls) - (hstep : step s (.bytes bs) = .ok (s', actions)) : - actions = [.write (encodeLs s.supported)] ∧ s'.phase = .negotiating ∧ s'.buffer = rest := by - simp only [step, hdec] at hstep - have hc := classifyListIds hrole hphase hls - rw [hc] at hstep - have ha := actListIds hstep - rcases ha with ⟨hacts, hphase', hbuf⟩ - constructor - · exact hacts - constructor - · rw [hphase'] - exact hphase - · exact hbuf - -/-- An unknown id draws the `na` frame and stays in the negotiation -(contract 2). -/ -theorem stepResponderUnknownStays {s : State} {bs : List Nat} {s' : State} - {actions : List Action} {m : Msg} {rest : List Nat} {id : Libp2p.Core.ProtocolId} - (hrole : s.role = .responder) (hphase : s.phase = .negotiating) - (hdec : decodeMsg? (s.buffer ++ bs) = .ok (m, rest)) - (htext : m.text = id.toString) (hnot : id ∉ s.supported) - (hheader : m.text ≠ protocolId.toString) - (hreply : Reply.ofMsg? m = none) - (hstep : step s (.bytes bs) = .ok (s', actions)) : - actions = [.write (encodeMsg (Reply.toMsg .na))] ∧ s'.phase = .negotiating ∧ s'.buffer = rest := by - have hnot' : (⟨m.text⟩ : Libp2p.Core.ProtocolId) ∉ s.supported := by - intro hm - apply hnot - simpa [htext] using hm - simp only [step, hdec] at hstep - have hc := classifyRefuse hrole hphase hnot' hheader hreply - rw [hc] at hstep - have ha := actRefuse hstep - rcases ha with ⟨hacts, hphase', hbuf⟩ - constructor - · exact hacts - constructor - · rw [hphase'] - exact hphase - · exact hbuf - -/-- Conservation: the bytes the accepting initiator consumed, plus the -leftover, equal the bytes fed in (contract 5). -/ -theorem stepSelectConservation {s : State} {bs : List Nat} {s' : State} - {p : Libp2p.Core.ProtocolId} {rest r : List Nat} {m : Msg} - (hrole : s.role = .initiator) - (hdec : decodeMsg? (s.buffer ++ bs) = .ok (m, r)) - (hstep : step s (.bytes bs) = .ok (s', [Action.select p rest])) : - s.buffer ++ bs = encodeText p.toString ++ rest := by - simp only [step, hdec] at hstep - have hmem : Action.select p rest ∈ [Action.select p rest] := by simp - have hv := actSelect hstep hmem - rcases hv with hc | hc - · exact (classifyPickInitiator hrole hc).elim - · have hpt : p.toString = m.text := classifyAcceptText hc - rw [hc] at hstep - have hr : r = rest := by - unfold act at hstep - simp only [Except.ok.injEq, Prod.mk.injEq] at hstep - obtain ⟨-, hact⟩ := hstep - injection hact with hsel hnil - injection hsel with hp hrest - have hlaw : encodeText m.text ++ r = s.buffer ++ bs := - decodeMsgEncodeDecode (s.buffer ++ bs) m r hdec - rw [← hlaw] - rw [← hpt] - rw [hr] - -/-- A truncated buffer waits: no actions, no error, the phase holds, -and the buffer keeps the bytes (§3.1). -/ -theorem stepTruncatedWaits {s : State} {bs : List Nat} {s' : State} {actions : List Action} - (hdec : decodeMsg? (s.buffer ++ bs) = .error .truncated) - (hstep : step s (.bytes bs) = .ok (s', actions)) : - actions = [] ∧ s'.phase = s.phase ∧ s'.buffer = s.buffer ++ bs := by - simp only [step, hdec] at hstep - simp only [Except.ok.injEq, Prod.mk.injEq] at hstep - obtain ⟨hstate, hact⟩ := hstep - subst hact - rw [← hstate] - simp - -end Libp2p.Negotiate.Legacy \ No newline at end of file diff --git a/packages/libp2p/negotiate/Libp2p/Negotiate/Legacy/Upgrade.lean b/packages/libp2p/negotiate/Libp2p/Negotiate/Legacy/Upgrade.lean deleted file mode 100644 index 3587d59..0000000 --- a/packages/libp2p/negotiate/Libp2p/Negotiate/Legacy/Upgrade.lean +++ /dev/null @@ -1,639 +0,0 @@ -import Libp2p.Negotiate.Legacy.Step -import Libp2p.Negotiate.Legacy.Message -import Libp2p.Core -import Libp2p.Identity -import Streams.Conn -import Streams.BufReader -import Streams.Constants -import Conc -import Bytes - -/-! -# The upgrader - -`Upgrade` walks a raw `Connection` through security negotiation, the -security handshake, the identity check, and muxer negotiation into a -`MuxSession` (root §4.9, negotiate ARCHITECTURE, "Public surface"). The -pure part is `drive`: one fuel-bounded run over a byte cursor that -feeds the 1a core one `Event.bytes` call per complete frame. The IO -part is `negotiate`, one negotiation stage, and `upgradeWith`, the -three-stage pipeline under one `Conc.withinDeadline`. - -The driver owns the byte cursor. It never reads `s.buffer`; it resets -the buffer before every `step` call, and every unconsumed byte stays -in the `input` argument. The `leftover` field of the result is -authoritative. The shell passes the driver's own states and leftovers -back unchanged. - -The negotiation loop parks on the connection read until a selection, -an error, a close, or the deadline (root §3.3). No measure bounds that -wait, so the loop keeps the package's `repeat` form, the form of -`Conc.TaskSet.joinAll`, and stays outside the proved core. The pure -`drive` is total, and its theorems carry no IO obligation. - -The three stages thread the leftover bytes by contract 5 of the -package docs: the negotiation leftover seeds the handshake, the -secure connection delivers the handshake leftover through its reader, -and the muxer negotiation leftover reaches `Muxer.open`. --/ - -set_option autoImplicit false - -namespace Libp2p.Negotiate.Legacy - -open Bytes (DecodeError) - -/-! ## The driver -/ - -/-- The answer of one driver run. - -* `selected p consumed leftover writes`: a `step` selected `p`. - `consumed` is the concatenation of the encoded frames the run - consumed, the selected frame included. `leftover` is the byte suffix - after it. `writes` are the `Action.write` payloads, in order. -* `waiting s consumed leftover writes`: `decodeMsg?` reported - `truncated` on `leftover`. The caller reads more bytes and appends - them; it must not call `drive` again without new bytes. -* `running s consumed leftover writes`: the fuel ran out after a - complete frame. The caller calls `drive` again with `leftover`; it - must not read. -/ -inductive DriveResult where - | selected (p : Libp2p.Core.ProtocolId) (consumed leftover : List Nat) - (writes : List (List Nat)) - | waiting (s : State) (consumed leftover : List Nat) (writes : List (List Nat)) - | running (s : State) (consumed leftover : List Nat) (writes : List (List Nat)) - deriving Repr, DecidableEq - -namespace DriveResult - -/-- The frames the run consumed, as their encoded bytes. -/ -@[simp] def consumed : DriveResult → List Nat - | .selected _ c _ _ => c - | .waiting _ c _ _ => c - | .running _ c _ _ => c - -/-- The bytes the run did not consume. -/ -@[simp] def leftover : DriveResult → List Nat - | .selected _ _ l _ => l - | .waiting _ _ l _ => l - | .running _ _ l _ => l - -end DriveResult - -/-- The `Action.write` payloads of an action list, in order. -/ -def writesOf (actions : List Action) : List (List Nat) := - actions.filterMap fun - | .write bs => some bs - | .select .. => none - -/-- The protocol an action list selects, if it selects one. -/ -def selectOf : List Action → Option Libp2p.Core.ProtocolId - | [] => none - | .write _ :: rest => selectOf rest - | .select p _ :: _ => some p - -/-- One fuel unit consumes at most one complete frame. The -accumulators keep the consumed bytes and the writes of the run. -/ -private def driveAux : Nat → State → List Nat → List Nat → List (List Nat) → - Except Error DriveResult - | 0, s, input, consumed, writes => .ok (.running s consumed input writes) - | fuel + 1, s, input, consumed, writes => - match decodeMsg? input with - | .error .truncated => .ok (.waiting s consumed input writes) - | .error .malformed => .error (.decode .malformed) - | .ok (m, rest) => - match step { s with buffer := [] } (.bytes input) with - | .error e => .error e - | .ok (s', actions) => - match selectOf actions with - | some p => - .ok (.selected p (consumed ++ encodeMsg m) rest - (writes ++ writesOf actions)) - | none => - driveAux fuel s' rest (consumed ++ encodeMsg m) - (writes ++ writesOf actions) - -/-- Drive the pure core over a byte cursor. `s.buffer` is -unused: pass every unconsumed byte in `input`, and read the -`leftover` field of the result as authoritative. One fuel unit -consumes at most one frame. -/ -def drive (fuel : Nat) (s : State) (input : List Nat) : Except Error DriveResult := - driveAux fuel s input [] [] - -/-- The initiator's post-propose state and its one coalesced write. -The write holds the header and the first proposal back-to-back, the -lazy dialer's wire shape (contract 4 of the package docs). The state -carries the whole ordered list, so the core's `na` retry walks it. -An empty ordered list is a protocol error before any byte moves. -/ -def initiatorStart (ids : List Libp2p.Core.ProtocolId) : - Except Error (State × List (List Nat)) := - match ids with - | [] => .error .negotiationFailed - | p :: _ => - if ids.any (fun q => !fitsText q.toString) then .error .negotiationFailed - else - .ok ({ role := .initiator, supported := [], proposals := ids, buffer := [], - phase := .negotiating }, - [encodeMsg header ++ encodeText p.toString]) - -/-- The responder's start: validate the offered list, then send the header -at once. The spec lets either side send first, and the responder that waits -for the dialer's header stalls a dialer that also waits. The state starts in -`negotiating` because the header is already on the wire. -/ -def responderStart (ids : List Libp2p.Core.ProtocolId) : - Except Error (State × List (List Nat)) := - if ids.any (fun q => !fitsText q.toString) then .error .negotiationFailed - else if maxMsg < (lsBody ids).length then .error .negotiationFailed - else - .ok ({ role := .responder, supported := ids, proposals := [], buffer := [], - phase := .negotiating }, - [encodeMsg header]) - -/-! ## The error mappings -/ - -/-- The rendered form of a decode error. `Bytes.DecodeError` carries no -`ToString` instance, so the mapping renders it here. -/ -def renderDecodeError : Bytes.DecodeError → String - | .truncated => "truncated" - | .malformed => "malformed" - -/-- Contract 8's one mapping point: every negotiation error becomes -the `protocol` arm, with the detail rendered. -/ -def mapNegotiateError : Error → Libp2p.Core.Error - | .decode e => .protocol ("multistream decode: " ++ renderDecodeError e) - | .negotiationFailed => .protocol "multistream negotiation failed" - -/-- The identity gate. `none` is the listen path, which accepts any -peer. The dial path compares the derived id against the expectation -and answers `authentication` on a difference, before any muxer byte -moves (negotiate ARCHITECTURE, "Decisions"). -/ -def checkIdentity (expected : Option Libp2p.Identity.PeerId) - (remote : Libp2p.Identity.PeerId) : Except Libp2p.Core.Error Unit := - match expected with - | none => .ok () - | some e => - if e = remote then .ok () - else .error (.authentication "remote peer id differs from the expected peer id") - -/-- The race answer flattened. `Conc.withinDeadline` constructs -`.error .deadlineExpired`; the `.closed` arm is required by the match -and is unreachable from `withinDeadline`. -/ -def settleUpgrade {α : Type} : - Except Conc.Error (Except Libp2p.Core.Error α) → Except Libp2p.Core.Error α - | .ok (.ok a) => .ok a - | .ok (.error e) => .error e - | .error .deadlineExpired => .error (.timeout "upgrade deadline expired") - | .error .closed => .error (.closed "upgrade deadline select closed") - -/-- The answer of one negotiation stage: the selected id and the bytes -the stage over-read. `ByteArray` carries no `Repr` in Lean 4.34, so -the structure derives none; the suite compares the fields it needs. -/ -structure StageResult where - proto : Libp2p.Core.ProtocolId - leftover : ByteArray - -/-! ## The driver theorems -/ - -/-- The accumulator invariant behind the conservation law: the bytes -fed in equal the bytes consumed plus the final leftover, for every -arm of the result. The proof inducts on the fuel; -`decodeMsgEncodeDecode` reassociates each consumed frame. -/ -private theorem driveAux_conservation (fuel : Nat) : - ∀ (s : State) (input consumed : List Nat) (writes : List (List Nat)) - (r : DriveResult), - driveAux fuel s input consumed writes = .ok r → - consumed ++ input = r.consumed ++ r.leftover := by - induction fuel with - | zero => - intro s input consumed writes r h - simp only [driveAux, Except.ok.injEq] at h - subst h - simp - | succ fuel ih => - intro s input consumed writes r h - simp only [driveAux] at h - split at h - · simp only [Except.ok.injEq] at h - subst h - simp - · simp at h - · rename_i m rest hdec - split at h - · rename_i e - simp at h - · rename_i s' actions hstep - split at h - · rename_i p - simp only [Except.ok.injEq] at h - subst h - have hbs := decodeMsgEncodeDecode input m rest hdec - rw [← hbs] - simp [List.append_assoc] - · have hrec := ih s' rest (consumed ++ encodeMsg m) - (writes ++ writesOf actions) r h - have hbs := decodeMsgEncodeDecode input m rest hdec - rw [← hbs, ← List.append_assoc] - exact hrec - -/-- Every byte fed to the driver is consumed or left over. -/ -theorem driveConservation {fuel : Nat} {s : State} {input : List Nat} {r : DriveResult} - (h : drive fuel s input = .ok r) : input = r.consumed ++ r.leftover := by - have h' := driveAux_conservation fuel s input [] [] r h - simpa using h' - -/-- The selected arm: the bytes in equal the frames consumed plus -the leftover handed to the next stage (contract 5). -/ -theorem driveSelectedConservation {fuel : Nat} {s : State} {input : List Nat} - {p : Libp2p.Core.ProtocolId} {consumed leftover : List Nat} - {writes : List (List Nat)} - (h : drive fuel s input = .ok (.selected p consumed leftover writes)) : - input = consumed ++ leftover := by - simpa using driveConservation h - -/-- The truncated arm: every byte the run saw sits in the -consumed frames or in the waiting buffer, so the shell loses no byte -when it reads the next chunk. -/ -theorem driveWaitingConservation {fuel : Nat} {s s' : State} {input : List Nat} - {consumed leftover : List Nat} {writes : List (List Nat)} - (h : drive fuel s input = .ok (.waiting s' consumed leftover writes)) : - input = consumed ++ leftover := by - simpa using driveConservation h - -/-- The accumulator prefix invariant: a run appends the frames it -consumes to the accumulator it received. -/ -private theorem driveAux_consumed_prefix (fuel : Nat) : - ∀ (s : State) (input consumed : List Nat) (writes : List (List Nat)) - (r : DriveResult), - driveAux fuel s input consumed writes = .ok r → - ∃ tail : List Nat, r.consumed = consumed ++ tail := by - induction fuel with - | zero => - intro s input consumed writes r h - simp only [driveAux, Except.ok.injEq] at h - subst h - exact ⟨[], by simp⟩ - | succ fuel ih => - intro s input consumed writes r h - simp only [driveAux] at h - split at h - · simp only [Except.ok.injEq] at h - subst h - exact ⟨[], by simp⟩ - · simp at h - · rename_i m rest hdec - split at h - · rename_i e - simp at h - · rename_i s' actions hstep - split at h - · rename_i p - simp only [Except.ok.injEq] at h - subst h - exact ⟨encodeMsg m, rfl⟩ - · obtain ⟨tail, htail⟩ := - ih s' rest (consumed ++ encodeMsg m) (writes ++ writesOf actions) r h - exact ⟨encodeMsg m ++ tail, by rw [htail, List.append_assoc]⟩ - -/-- A frame the shell writes through `encodeMsg` decodes and appears -at the head of the consumed bytes. -/ -theorem driveConsumedHead {fuel : Nat} {s : State} {m : Msg} {rest : List Nat} - {r : DriveResult} - (h : drive (fuel + 1) s (encodeMsg m ++ rest) = .ok r) : - ∃ tail : List Nat, r.consumed = encodeMsg m ++ tail := by - have h' : driveAux (fuel + 1) s (encodeMsg m ++ rest) [] [] = .ok r := h - simp only [driveAux, encodeMsg, decodeMsgSuffix] at h' - split at h' - · rename_i e - simp at h' - · rename_i s' actions hstep - split at h' - · rename_i p - simp only [Except.ok.injEq] at h' - subst h' - exact ⟨[], by simp [encodeMsg]⟩ - · obtain ⟨tail, htail⟩ := - driveAux_consumed_prefix fuel s' rest ([] ++ encodeMsg m) (writesOf actions) r h' - exact ⟨tail, by simpa [encodeMsg, List.append_assoc] using htail⟩ - -/-- Two stages compose: bytes in equal the consumed bytes of both -stages plus the final leftover. The reusable form of the leftover -hand-over (contract 5). -/ -theorem handoverCompose {ε α β : Type} - {f : List Nat → Except ε (α × List Nat)} - {g : List Nat → Except ε (β × List Nat)} - {c₁ : α → List Nat} {c₂ : β → List Nat} - {bs₁ bs₂ l₁ l₂ : List Nat} {a₁ : α} {a₂ : β} - (law₁ : ∀ x a r, f x = .ok (a, r) → x = c₁ a ++ r) - (law₂ : ∀ x a r, g x = .ok (a, r) → x = c₂ a ++ r) - (h₁ : f bs₁ = .ok (a₁, l₁)) - (h₂ : g (l₁ ++ bs₂) = .ok (a₂, l₂)) : - bs₁ ++ bs₂ = c₁ a₁ ++ c₂ a₂ ++ l₂ := by - rw [law₁ bs₁ a₁ l₁ h₁, List.append_assoc, law₂ (l₁ ++ bs₂) a₂ l₂ h₂] - simp only [List.append_assoc] - -/-- The pure form of one hand-over: when a second drive starts from -the first drive's leftover, the bytes fed in equal the bytes both -drives consumed plus the final leftover. -/ -theorem driveTwostage {fuel₁ fuel₂ : Nat} {s₁ s₂ : State} - {input₁ input₂ : List Nat} {p₁ p₂ : Libp2p.Core.ProtocolId} - {c₁ c₂ l₁ l₂ : List Nat} {w₁ w₂ : List (List Nat)} - (h₁ : drive fuel₁ s₁ input₁ = .ok (.selected p₁ c₁ l₁ w₁)) - (h₂ : drive fuel₂ s₂ (l₁ ++ input₂) = .ok (.selected p₂ c₂ l₂ w₂)) : - input₁ ++ input₂ = c₁ ++ c₂ ++ l₂ := by - have h1 := driveSelectedConservation h₁ - have h2 := driveSelectedConservation h₂ - rw [h1, List.append_assoc, h2] - simp only [List.append_assoc] - -/-- The shell's manual first write equals the core's propose step for -a singleton list. The theorem does not cover a longer list. -/ -theorem initiatorStartPropose {p : Libp2p.Core.ProtocolId} - (hfits : fitsText p.toString = true) : - initiatorStart [p] = - (step { role := .initiator, supported := [], proposals := [], buffer := [], - phase := .headerPending } (.propose p)).map - (fun sa => (sa.1, writesOf sa.2)) := by - have hany : ([p].any (fun q => !fitsText q.toString)) = false := by - simp [List.any, hfits] - simp only [initiatorStart, hany] - rfl - -/-- An empty ordered list is a protocol error before any byte is -written. -/ -theorem initiatorStartNil : initiatorStart [] = .error .negotiationFailed := rfl - -/-- The one mapping point of contract 8: every `Negotiate.Error` -becomes the `protocol` arm, with the detail rendered. -/ -theorem mapNegotiateErrorSpec {e : Bytes.DecodeError} : - mapNegotiateError (.decode e) = - .protocol ("multistream decode: " ++ renderDecodeError e) ∧ - mapNegotiateError .negotiationFailed = - .protocol "multistream negotiation failed" := by - simp [mapNegotiateError] - -/-- A differing derived id is authentication, before any muxer -byte moves. -/ -theorem checkIdentityReject {expected remote : Libp2p.Identity.PeerId} - (h : expected ≠ remote) : - checkIdentity (some expected) remote = - .error (.authentication "remote peer id differs from the expected peer id") := by - simp [checkIdentity, h] - -/-- The dial to the expected peer passes. -/ -theorem checkIdentityAccept {e : Libp2p.Identity.PeerId} : - checkIdentity (some e) e = .ok () := by - simp [checkIdentity] - -/-- The listen path accepts any peer. -/ -theorem checkIdentityNone {r : Libp2p.Identity.PeerId} : - checkIdentity none r = .ok () := rfl - -/-- The deadline wiring of contract 6: the race's expiry is the -`timeout` arm, and the seam errors pass through untouched. The -`.closed` arm is required by the match over `Conc.Error`. -/ -theorem settleUpgradeSpec {α : Type} {e : Libp2p.Core.Error} {a : α} : - settleUpgrade (α := α) (.error .deadlineExpired) = - .error (.timeout "upgrade deadline expired") ∧ - settleUpgrade (α := α) (.error .closed) = - .error (.closed "upgrade deadline select closed") ∧ - settleUpgrade (α := α) (.ok (.error e)) = .error e ∧ - settleUpgrade (.ok (.ok a)) = .ok a := by - simp [settleUpgrade] - -/-! ## The IO shell -/ - -/-- Write one frame, and turn a fault of the connection into the -`closed` arm. `Connection.write` throws `IO.Error`, which is outside -this shell's `Core.Error` channel. -/ -private def writeFrame (conn : Streams.Connection) (b : ByteArray) : - ExceptT Libp2p.Core.Error Conc.Async Unit := do - let ok ← (try - conn.write b - pure true - catch _ => pure false : Conc.Async Bool) - if ok then pure () else throw (.closed "connection write failed during the upgrade") - -/-- Write each frame payload through the connection, in order. A -failed write is the `closed` arm: the connection is gone. -/ -private def sendWrites (conn : Streams.Connection) (ws : List (List Nat)) : - ExceptT Libp2p.Core.Error Conc.Async Unit := do - for w in ws do - writeFrame conn (Bytes.ofNatList w) - -/-- The negotiation loop. It parks on `reader.read` only when `drive` -reports `waiting`; a `running` answer costs no read, because the driver -has not yet tried to decode `leftover`. `reader.read` returns the next -reader, and the loop keeps it for the next read. The read parks for as -long as the peer stays silent, so no measure bounds the loop (root -§3.3). The `repeat` form keeps the definition free of a `partial` -marker, like `Conc.TaskSet.joinAll`. -/ -private def negotiateLoop (reader : Streams.BufReader) - (conn : Streams.Connection) (s : State) (input : List Nat) : - ExceptT Libp2p.Core.Error Conc.Async StageResult := do - let mut s := s - let mut input := input - let mut reader := reader - repeat - match drive 1 s input with - | .error e => throw (mapNegotiateError e) - | .ok (.selected p _ leftover writes) => do - sendWrites conn writes - return { proto := p, leftover := Bytes.ofNatList leftover } - | .ok (.waiting s' _ leftover writes) => do - sendWrites conn writes - match ← reader.read with - | none => throw (.closed "multistream negotiation ended before a selection") - | some (chunk, reader') => - s := s' - input := leftover ++ Bytes.toNatList chunk - reader := reader' - | .ok (.running s' _ leftover writes) => do - sendWrites conn writes - s := s' - input := leftover - -- Every arm above returns or loops, so the `repeat` never falls - -- through. The tail gives the `do` block its result type without an - -- `Inhabited`. - throw (.closed "the negotiation loop ended without an answer") - -/-- One negotiation stage. Both roles write before the loop: the -initiator writes the coalesced header plus first proposal, and the -responder writes its header. The seed is the leftover of the previous -stage. The end of the stream mid-negotiation is `Core.Error.closed`; a -`Negotiate.Error` maps once through `mapNegotiateError`. -/ -def negotiate (role : Libp2p.Core.Role) (reader : Streams.BufReader) - (conn : Streams.Connection) (seed : ByteArray) - (ids : List Libp2p.Core.ProtocolId) : - ExceptT Libp2p.Core.Error Conc.Async StageResult := do - match role with - | .initiator => - match initiatorStart ids with - | .error e => throw (mapNegotiateError e) - | .ok (s, first) => do - sendWrites conn first - negotiateLoop reader conn s (Bytes.toNatList seed) - | .responder => - match responderStart ids with - | .error e => throw (mapNegotiateError e) - | .ok (s, first) => do - sendWrites conn first - negotiateLoop reader conn s (Bytes.toNatList seed) - -/-- An upgraded session and its secure connection completion. -/ -structure Upgraded where - session : Libp2p.Core.MuxSession - remotePeer : Libp2p.Identity.PeerId - join : ExceptT Libp2p.Core.Error Conc.Async Unit - -/-- The result of the post-handshake admission checkpoint. -/ -inductive UpgradeResult (ρ : Type) where - | upgraded (value : Upgraded) - | refused (reason : ρ) - -private def upgradeCore {ρ : Type} (role : Libp2p.Core.Role) (conn : Streams.Connection) - (leftover : ByteArray) (expected : Option Libp2p.Identity.PeerId) - (admit : Libp2p.Identity.PeerId → BaseIO (Option ρ)) - (secs : List Libp2p.Core.SecurityProtocol) (muxers : List Libp2p.Core.Muxer) : - Conc.Async (Except Libp2p.Core.Error (UpgradeResult ρ)) := do - let reader := Streams.BufReader.new conn - match ← negotiate role reader conn leftover (secs.map (·.id)) with - | .error e => return .error e - | .ok stage => - match secs.find? (fun s => decide (s.id = stage.proto)) with - | none => return .error (.protocol "selected security protocol is not configured") - | some sec => - match ← sec.handshake role conn stage.leftover with - | .error e => return .error e - | .ok sc => - match checkIdentity expected sc.remotePeer with - | .error e => - sc.toConnection.closeQuietly - discard sc.join.run - return .error e - | .ok () => - if let some reason ← (admit sc.remotePeer : BaseIO (Option ρ)) then - sc.toConnection.closeQuietly - discard sc.join.run - return .ok (.refused reason) - let muxReader := Streams.BufReader.new sc.toConnection - match ← negotiate role muxReader sc.toConnection ByteArray.empty (muxers.map (·.id)) with - | .error e => - sc.toConnection.closeQuietly - discard sc.join.run - return .error e - | .ok stage2 => - match muxers.find? (fun m => decide (m.id = stage2.proto)) with - | none => - sc.toConnection.closeQuietly - discard sc.join.run - return .error (.protocol "selected muxer is not configured") - | some mux => - match ← mux.«open» role sc.toConnection stage2.leftover with - | .error e => - sc.toConnection.closeQuietly - discard sc.join.run - return .error e - | .ok session => - let join : ExceptT Libp2p.Core.Error Conc.Async Unit := ExceptT.mk do - let muxResult ← session.join.run - let secureResult ← sc.join.run - return match muxResult, secureResult with - | .error e, _ => .error e - | _, .error e => .error e - | .ok (), .ok () => .ok () - return .ok (.upgraded { session, remotePeer := sc.remotePeer, join }) - -/-- Whether the deadline boundary already closed the connection. -`withinDeadline` returns an error value only on expiry, after the hook -ran. -/ -private def boundaryClosed {α : Type} : Except Conc.Error α → Bool - | .error _ => true - | .ok _ => false - -/-- Close a session that finished after the timer won. No caller -receives that session. -/ -private def releaseLate {ρ : Type} (r : Except Libp2p.Core.Error (UpgradeResult ρ)) : BaseIO Unit := - match r with - | .ok (.upgraded value) => discard value.session.close.toBaseIO - | _ => pure () - -/-- The upgrade under one `Conc.withinDeadline`. The expiry closes -the connection once at the boundary (root §3.5), and the race's -`deadlineExpired` maps to the `timeout` arm through `settleUpgrade`. -The clock is a seam, so the deadline test can advance it. -/ - -def upgradeAdmit {ρ : Type} (clock : Conc.Clock) (role : Libp2p.Core.Role) - (conn : Streams.Connection) (leftover : ByteArray) - (expected : Option Libp2p.Identity.PeerId) - (admit : Libp2p.Identity.PeerId → BaseIO (Option ρ)) - (secs : List Libp2p.Core.SecurityProtocol) (muxers : List Libp2p.Core.Muxer) : - ExceptT Libp2p.Core.Error Conc.Async (UpgradeResult ρ) := do - let answer ← Conc.withinDeadline clock Streams.unauthDeadlineMs - (Streams.Connection.closeQuietly conn) - (upgradeCore role conn leftover expected admit secs muxers) - releaseLate - match settleUpgrade answer with - | .ok pair => return pair - | .error e => do - -- An error from the boundary means the expiry hook closed the - -- connection. A failure inside the work leaves it open. - if boundaryClosed answer then throw e - else do - Streams.Connection.closeQuietly conn - throw e - -/-- Upgrade with a clock and admit every authenticated peer. -/ -def upgradeWith (clock : Conc.Clock) (role : Libp2p.Core.Role) - (conn : Streams.Connection) (leftover : ByteArray) - (expected : Option Libp2p.Identity.PeerId) - (secs : List Libp2p.Core.SecurityProtocol) (muxers : List Libp2p.Core.Muxer) : - ExceptT Libp2p.Core.Error Conc.Async (Libp2p.Core.MuxSession × Libp2p.Identity.PeerId) := do - match ← upgradeAdmit clock role conn leftover expected (fun _ => pure (none : Option Empty)) secs muxers with - | .upgraded value => return (value.session, value.remotePeer) - | .refused reason => nomatch reason - -/-- The frozen entry of the package docs. The clock is fixed to the -system clock here and stays a seam in `upgradeWith` for the deadline -test. -/ -def upgrade (role : Libp2p.Core.Role) (conn : Streams.Connection) - (leftover : ByteArray) (expected : Option Libp2p.Identity.PeerId) - (secs : List Libp2p.Core.SecurityProtocol) (muxers : List Libp2p.Core.Muxer) : - ExceptT Libp2p.Core.Error Conc.Async (Libp2p.Core.MuxSession × Libp2p.Identity.PeerId) := - upgradeWith Conc.Clock.system role conn leftover expected secs muxers - -/-- Negotiate a protocol on a muxed stream and retain any extra bytes. -/ -def negotiateStream (role : Libp2p.Core.Role) (stream : Streams.Stream) - (ids : List Libp2p.Core.ProtocolId) : - ExceptT Libp2p.Core.Error Conc.Async (Libp2p.Core.ProtocolId × Streams.Stream) := do - let (initial, writes) ← match role with - | .initiator => (initiatorStart ids).mapError mapNegotiateError - | .responder => (responderStart ids).mapError mapNegotiateError - let mut state := initial - let mut input : List Nat := [] - for bytes in writes do - if let .error e ← stream.write (Bytes.ofNatList bytes) then - throw (.closed s!"stream negotiation write failed: {repr e}") - repeat - match drive 1 state input with - | .error e => throw (mapNegotiateError e) - | .ok result => - let (next, rest, writes, selected) := match result with - | .selected p _ rest writes => (state, rest, writes, some p) - | .waiting s _ rest writes => (s, rest, writes, none) - | .running s _ rest writes => (s, rest, writes, none) - for bytes in writes do - if let .error e ← stream.write (Bytes.ofNatList bytes) then - throw (.closed s!"stream negotiation write failed: {repr e}") - if let some p := selected then - let s ← Streams.Stream.prependBytes stream (Bytes.ofNatList rest) - return (p, s) - state := next - input := rest - if result matches .waiting .. then - match ← stream.read with - | .error e => throw (.closed s!"stream negotiation read failed: {repr e}") - | .ok none => throw (.closed "stream ended during negotiation") - | .ok (some bytes) => input := input ++ Bytes.toNatList bytes - throw (.closed "stream negotiation ended without a selection") - -end Libp2p.Negotiate.Legacy \ No newline at end of file diff --git a/packages/libp2p/protocols/Libp2p/Protocols/Mock/HostStream.lean b/packages/libp2p/protocols/Libp2p/Protocols/Mock/HostStream.lean deleted file mode 100644 index dc68a99..0000000 --- a/packages/libp2p/protocols/Libp2p/Protocols/Mock/HostStream.lean +++ /dev/null @@ -1,144 +0,0 @@ -import Conc -import Streams -import Libp2p.Core.Handler -import Libp2p.Core.Mux -import Libp2p.Protocols.StreamError - -/-! -# Host stream mock - -`Libp2p.Host` (W5.6) merged, but it still runs its upgrade through a -mock of `libp2p/negotiate` (issue #70), so no host opens a session yet. -The protocols docs have the req/resp client open its stream through -the host, with `newStream`, and serve through `setHandler` over the -registry of §3.10; go-libp2p's outbound call is `NewStream` -(`vendor/go-libp2p/p2p/host/basic/basic_host.go`, line 441). This -mock fixes minimal signatures for both sides over one muxer session of -the real `MuxSession` seam (RULES.md, "Working ahead of the tree"). - -## What it leaves out - -* Dialing: the mock holds one session, so `newStream` needs no peer. -* Negotiation (W5.5): the opener writes the first id it asks for, with a - one-byte length (an id of at most 255 bytes), and the listener looks - the id up in its registry. An id it does not serve resets the stream. - The bytes the listener read past the id go in front of the stream - with `Stream.prependBytes` (§4.7), as the negotiation hands them on. -* Failure reporting: a handler's failure goes to `failures`, the place - the host's `onFailure` will take. - -## Tasks and parks - -`serve` accepts in a loop and runs each stream in a task of a set it -owns, then joins the set when the session ends. The accept parks on the -session, which `close` ends. `close` closes the session, and `join` -waits for the serve loop and the session loop. - -## Integration - -Delete this module. Use the host's `newStream` for the peer, bound to -the peer id, in place of `Host.newStream`, and the host's `setHandler` -in place of `Host.setHandler`. --/ - -set_option autoImplicit false - -namespace Libp2p.Protocols.Mock.HostStream - -open Conc Streams -open Libp2p.Core (ProtocolId StreamContext StreamHandler MuxSession) - -/-- One end of a peer connection, with its handler registry. -/ -structure Host where - /-- The mock session to the peer. -/ - session : MuxSession - /-- The context every served stream of the session gets: the peer and - its address, which the real host takes from the upgrade. -/ - ctx : StreamContext - /-- The registry, by protocol id; changed only through `update`. -/ - handlers : OnceRef (List StreamHandler) - /-- The handler failures, newest first. -/ - failures : OnceRef (List Libp2p.Core.Error) - /-- The tasks of the served streams and the serve loop. -/ - owner : TaskSet - -/-- A host over a session to the peer that `ctx` names. -/ -def Host.new (session : MuxSession) (ctx : StreamContext) : BaseIO Host := do - return ⟨session, ctx, ← OnceRef.new [], ← OnceRef.new [], ← TaskSet.new⟩ - -/-- Store one handler under its id; a second store replaces the first. -/ -def Host.setHandler (h : Host) (sh : StreamHandler) : BaseIO Unit := - h.handlers.update fun hs => sh :: hs.filter (·.id != sh.id) - -/-- The header the opener writes: the id with a one-byte length. -/ -def idHeader (id : ProtocolId) : Option ByteArray := - let b := id.toString.toUTF8 - if b.size < 256 then some (ByteArray.mk #[b.size.toUInt8] ++ b) else none - -/-- Open a stream to the peer for the first id of `protocols`. -/ -def Host.newStream (h : Host) (protocols : List ProtocolId) : - Async (Except Libp2p.Core.Error (ProtocolId × Stream)) := do - let some id := protocols.head? | return .error (.protocol "newStream: no protocol id") - let some header := idHeader id | return .error (.protocol "newStream: an id over 255 bytes") - match ← (h.session.openStream id).run with - | .error e => return .error e - | .ok s => - match ← s.write header with - | .error e => s.reset; return .error (mapStreamError e) - | .ok () => return .ok (id, s) - -/-- Read the id header from `s`: the id and the bytes after it. `none` -when the stream ends or fails first. The read parks on the stream, which -the session close or a reset wakes; the loop is `partial` for the same -reason as every stream read loop. -/ -partial def readHeader (s : Stream) (buf : ByteArray) : Async (Option (String × ByteArray)) := do - if h : 0 < buf.size then - let len := (buf.get 0 h).toNat - if 1 + len ≤ buf.size then - return (String.fromUTF8? (buf.extract 1 (1 + len))).map fun id => - (id, buf.extract (1 + len) buf.size) - match ← s.read with - | .ok (some b) => readHeader s (buf ++ b) - | _ => return none - -/-- Serve one inbound stream through the registry. -/ -def serveStream (h : Host) (s : Stream) : Async Unit := do - let some (id, rest) ← readHeader s ByteArray.empty | s.reset - let hs ← (h.handlers.get : BaseIO _) - match hs.find? (·.id.toString == id) with - | none => s.reset - | some sh => - let s ← s.prependBytes rest - match ← (sh.handle h.ctx s).run with - | .ok () => pure () - | .error e => - h.failures.update (e :: ·) - s.reset - -/-- The serve loop: accept until the session ends, one task per stream. -`partial`: the peer may open streams forever; the session close ends the -accept with `none`. -/ -partial def serveLoop (h : Host) (streams : TaskSet) : Async Unit := do - -- Park: a new stream of the peer wakes it, and so does the close. - match ← h.session.accept with - | none => streams.joinAll - | some s => - streams.spawn (serveStream h s) - serveLoop h streams - -/-- Start serving inbound streams. -/ -def Host.serve (h : Host) : BaseIO Unit := do - let streams ← TaskSet.new - h.owner.spawn (serveLoop h streams) - -/-- Close the session. -/ -def Host.close (h : Host) : IO Unit := h.session.close - -/-- Wait for the serve loop, its streams, and the session loop. The -session of `Mock.Muxer.muxer` always joins with `.ok`, so the answer -goes unread. -/ -def Host.join (h : Host) : Async Unit := do - h.owner.joinAll - discard h.session.join.run - -end Libp2p.Protocols.Mock.HostStream diff --git a/packages/libp2p/protocols/Libp2p/Protocols/Mock/Muxer.lean b/packages/libp2p/protocols/Libp2p/Protocols/Mock/Muxer.lean deleted file mode 100644 index a451b43..0000000 --- a/packages/libp2p/protocols/Libp2p/Protocols/Mock/Muxer.lean +++ /dev/null @@ -1,222 +0,0 @@ -import Std.Data.HashMap -import Conc -import Streams -import Libp2p.Core.Mux - -/-! -# Muxer mock - -The muxer values of `libp2p/mux` took the real core seams while this -branch ran (PR #84, issue #77 closed), so the swap of this mock is due -now (issue #78). Until it lands, this module stands in with one value, -`muxer`, of the real `Muxer` seam of `libp2p/core`. It runs over any -`Connection`, so the req/resp suite runs over the sim transport -(PRINCIPLES 1.4; RULES.md, "Working ahead of the tree"). - -## The wire of the mock - -One frame shape, and nothing else: - -``` - -``` - -The kind is data (0), close (1), or reset (2). The 2-byte length caps a -frame at 65535 bytes, and the reader takes no more (§5.1). The dialer -numbers its streams odd and the listener even, so the two sides never -pick one id. The first frame of a new id from the peer opens the -stream, and `accept` hands it out. A close ends the peer's read -direction only (§5.3). A reset fails both sides, also after a close. -The end of the session fails every open stream with `.closed`. - -The mock enforces no flow control: the inboxes are unbounded, and the -yamux window of W5.4 is what bounds them in production. - -## Tasks and parks - -One reader task per session, in a `TaskSet` the session owns; `join` -joins it. The reader parks on `Connection.read`, which `close` wakes. -Writers take a one-token channel around each frame write, so two -streams never interleave the bytes of their frames; the token's park -wakes when the writer before it sends the token back. - -## Integration - -Delete this module, and use `Mux.test` or `Mux.yamux` where the suite -takes `muxer`. The suite needs no other change: it uses the seam fields -only. --/ - -set_option autoImplicit false - -namespace Libp2p.Protocols.Mock.Muxer - -open Conc Streams -open Libp2p.Core (Role MuxSession Muxer) - -/-! ## Frames -/ - -/-- The header of a frame: id, kind, length. -/ -def headerLen : Nat := 7 - -/-- The largest payload of one frame. -/ -def maxPayload : Nat := 2 ^ 16 - 1 - -/-- One frame on the wire. -/ -def frameB (sid kind : Nat) (payload : ByteArray) : ByteArray := - ByteArray.mk #[(sid / 2 ^ 24).toUInt8, (sid / 2 ^ 16).toUInt8, (sid / 2 ^ 8).toUInt8, - sid.toUInt8, kind.toUInt8, (payload.size / 256).toUInt8, payload.size.toUInt8] ++ payload - -/-- The frames of one write: the bytes in pieces of at most `maxPayload`. -/ -def dataFramesB (sid : Nat) (bytes : ByteArray) : ByteArray := Id.run do - let mut out := ByteArray.empty - let pieces := (bytes.size + maxPayload - 1) / maxPayload - for i in [0:pieces] do - out := out ++ frameB sid 0 (bytes.extract (i * maxPayload) (i * maxPayload + maxPayload)) - return out - -/-- One frame at the front of `b`: the id, the kind, the payload, and the -bytes it took; `none` while the frame is not whole. -/ -def readFrame? (b : ByteArray) : Option (Nat × Nat × ByteArray × Nat) := - if b.size < headerLen then none - else - let byte (i : Nat) : Nat := (b.get! i).toNat - let sid := ((byte 0 * 256 + byte 1) * 256 + byte 2) * 256 + byte 3 - let len := byte 5 * 256 + byte 6 - if b.size < headerLen + len then none - else some (sid, byte 4, b.extract headerLen (headerLen + len), headerLen + len) - -/-! ## The session -/ - -/-- The parts of one stream that the reader reaches. -/ -structure Entry where - /-- The chunks the reader delivers. -/ - inbox : Chan (Except Streams.Error ByteArray) - /-- Set by a reset, so a read after the close of the inbox fails. -/ - rst : OnceRef (Option Streams.Error) - -/-- What the streams and the reader of one session share. -/ -structure Shared where - conn : Connection - /-- The streams by id. Changed only through `update` and `updateGet`. -/ - streams : OnceRef (Std.HashMap Nat Entry) - /-- The next id this side opens. -/ - next : OnceRef Nat - /-- The streams the peer opened, for `accept`. -/ - incoming : Chan Stream - /-- The write token: one frame write at a time. -/ - token : Chan Unit - -/-- Write one frame run under the token. A failed write answers `false`. -/ -def Shared.send (sh : Shared) (bytes : ByteArray) : Async Bool := do - -- Park: the writer before this one sends the token back; a session - -- close closes the token channel and wakes this with `none`. - let some () ← sh.token.recv | return false - let ok ← try sh.conn.write bytes; pure true catch _ => pure false - discard <| sh.token.trySend () - return ok - -/-- A fresh entry. -/ -def newEntry : BaseIO Entry := do - return ⟨← Chan.newUnbounded, ← OnceRef.new none⟩ - -/-- The stream of id `sid` over the session. -/ -def mkStream (sh : Shared) (sid : Nat) (e : Entry) : BaseIO Stream := do - let completions ← Completions.new - Stream.ofPort { - inbox := e.inbox - endedBy := e.rst.get - completions - postWrite := fun tag bytes => do - let ok ← sh.send (dataFramesB sid bytes) - completions.resolve tag (if ok then .ok () else .error .closed) - return .ok () - postConsumed := fun _ => return .ok () - postClose := do - discard <| sh.send (frameB sid 1 ByteArray.empty) - return .ok () - postReset := do - e.rst.set (some .reset) - discard <| e.inbox.trySend (.error .reset) - e.inbox.close - sh.streams.update (·.erase sid) - discard <| sh.send (frameB sid 2 ByteArray.empty) - return .ok () } - -/-- Route one frame from the peer. A new id of the peer's parity opens a -stream; a frame for any other unknown id is dropped. -/ -def route (sh : Shared) (peerParity : Nat) (sid kind : Nat) (payload : ByteArray) : Async Unit := do - let entry? ← (sh.streams.get : BaseIO _) - let e ← match entry?.get? sid with - | some e => pure (some e) - | none => - if sid % 2 = peerParity ∧ kind ≠ 2 then - let e ← newEntry - sh.streams.update (·.insert sid e) - discard <| sh.incoming.trySend (← mkStream sh sid e) - pure (some e) - else pure none - let some e := e | return - match kind with - | 0 => discard <| e.inbox.trySend (.ok payload) - | 1 => e.inbox.close - | _ => - e.rst.set (some .reset) - discard <| e.inbox.trySend (.error .reset) - e.inbox.close - sh.streams.update (·.erase sid) - -/-- The reader loop. `partial`: a peer may send frames forever; the -session's close ends the connection read with `none`, which ends the -loop. At the end every inbox and `incoming` close, so each parked read -and accept wakes. -/ -partial def readLoop (sh : Shared) (peerParity : Nat) (buf : ByteArray) : Async Unit := do - match readFrame? buf with - | some (sid, kind, payload, k) => - route sh peerParity sid kind payload - readLoop sh peerParity (buf.extract k buf.size) - | none => - -- Park: the peer's bytes wake it, and so does `close` of the connection. - match ← sh.conn.read with - | some b => readLoop sh peerParity (buf ++ b) - | none => - -- The session ended under the open streams, so a read after this - -- answers `.closed`, the failure §5.2 keeps apart from an end. - for (_, e) in (← (sh.streams.get : BaseIO _)).toList do - e.rst.update fun r => r <|> some .closed - e.inbox.close - sh.incoming.close - sh.token.close - -/-- Run a mock session over `conn`. -/ -def openSession (role : Role) (conn : Connection) (leftover : ByteArray) : - ExceptT Libp2p.Core.Error Async MuxSession := do - let token : Chan Unit ← Chan.new 1 - discard <| token.trySend () - let sh : Shared := { - conn - streams := ← OnceRef.new {} - next := ← OnceRef.new (if role = .initiator then 1 else 2) - incoming := ← Chan.newUnbounded - token } - let owner ← TaskSet.new - let peerParity := if role = .initiator then 0 else 1 - owner.spawn (readLoop sh peerParity leftover) - return { - -- The mock wire carries no protocol label, so the id goes unread, - -- as yamux drops it (the core `MuxSession.openStream` doc). - openStream := fun _ => do - let sid ← (sh.next.updateGet fun n => (n, n + 2) : BaseIO Nat) - let e ← newEntry - sh.streams.update (·.insert sid e) - mkStream sh sid e - accept := sh.incoming.recv - close := sh.conn.close - -- The reader loop has no failure of its own, so the join answers - -- `.ok` once the loop ends. - join := ExceptT.lift owner.joinAll } - -/-- The mock muxer value. -/ -def muxer : Muxer := ⟨⟨"/leanp2p-mock-mux/1.0.0"⟩, openSession⟩ - -end Libp2p.Protocols.Mock.Muxer diff --git a/packages/libp2p/protocols/docs/ARCHITECTURE.md b/packages/libp2p/protocols/docs/ARCHITECTURE.md index 6cd6503..8cdcbb1 100644 --- a/packages/libp2p/protocols/docs/ARCHITECTURE.md +++ b/packages/libp2p/protocols/docs/ARCHITECTURE.md @@ -28,7 +28,6 @@ group is §7, row 10). The spike has no req/resp or gossipsub code. | `Libp2p/Protocols/ReqResp/Exchange.lean` | the shell over the core: the requester `request` and `requestPeer`, the responder `serve`, the handler value the registry stores, and the deadlines | | `Libp2p/Protocols/StreamError.lean` | `mapStreamError`, the one map from a stream error into `Core.Error` that the req/resp handler uses | | `Libp2p/Protocols/Mock/Core.lean` | the mock of W5.2 that identify and ping still type over, until their swap (issue #55) | -| `Libp2p/Protocols/Mock/{Muxer,HostStream}.lean` | the test values that stand in for a muxer value of W5.4 and the host of W5.6, over the real core seams; the muxer swap is due (`libp2p/mux` took the real core in PR #84, issue #77 closed, the swap owes on issue #78), and the host swap waits for the host's negotiate swap (issue #70) | | `Libp2p/Protocols/Gossipsub/Rpc.lean` | the RPC message with its `ProtoMessage` instance (§3.2) | | `Libp2p/Protocols/Gossipsub/Core.lean` | `State`, `Event`, `Action`, `ProtocolError`, and the total `step` that holds mesh, scoring, and heartbeat (§3.3) | | `Libp2p/Protocols/Gossipsub/Node.lean` | the loop task over the core, and the publish and subscribe surface at the seam level | @@ -96,9 +95,7 @@ package; §4.9 names no echo, so no module line covers it. stream to the peer and negotiating the id is the host's job (W5.6), as go-libp2p's `NewStream` does it (`vendor/go-libp2p/p2p/host/basic/basic_host.go`, line 441). The host - names it `newStream`. The host still upgrades through a mock of - `libp2p/negotiate` (issue #70), so the suite runs against a mock of - the call. `requestPeer` takes the call as an argument, with a slot + names it `newStream`. `requestPeer` takes the call as an argument, with a slot of the peer's `Limiter`, and runs `request` on the stream the call returns. `request` needs no host type: it writes the request, half-closes, and hands each response chunk to a callback @@ -264,16 +261,13 @@ Both spike programs call `Ping.ping`. forbids the request and a wait would park on another request's deadline. The slot goes back on every path. The responder's side of the rule, "MAY penalise", is scoring, and stays with the client. -- **The suite runs over the sim through two mocks.** The mock muxer is - one value of the `Muxer` seam of `libp2p/core` over a sim - `Connection`, and the host stream mock fixes minimal signatures for - the outbound stream call (go-libp2p's `NewStream`, - `vendor/go-libp2p/p2p/host/basic/basic_host.go`, line 441) and for - `setHandler`. Both type over the real core. `libp2p/mux` landed its - values over the real core (PR #84, issue #77 closed), so the muxer - swap is due (issue #78). The host swap waits for the host's - negotiate swap (issue #70). The suite then runs over yamux and the - host. +- **The suite runs between two hosts over the sim.** `Tests/HostPair.lean` + builds two real hosts on the test values of root §3.1: the sim + transport, the fixed-key Noise value, and a test clock, with yamux. The + requester opens its stream through the host's `newStream` (go-libp2p's + `NewStream`, `vendor/go-libp2p/p2p/host/basic/basic_host.go`, line + 441). The responder registers through `setHandler`. A host failure + that wraps a seam error reaches `requestPeer` as that error. ## What the package must not depend on diff --git a/packages/libp2p/protocols/docs/PLAN.md b/packages/libp2p/protocols/docs/PLAN.md index 38dfa42..bd5e383 100644 --- a/packages/libp2p/protocols/docs/PLAN.md +++ b/packages/libp2p/protocols/docs/PLAN.md @@ -94,11 +94,15 @@ in the codec group wrapping upstream `google/snappy`. - The concurrency limit: two requests with one id take a slot each, a third is refused, another id has its own slots, and a request gives its slot back when it ends (the bound of `Table.take` is a theorem). -- Over the sim transport, through the mock muxer and the host stream - mock (PRINCIPLES 1.4): a request through the host gets its chunks, - with whole and with split writes; an id the peer does not serve - resets the stream; a silent peer fails the request at its deadline; - a connection closed under a parked read ends the request. +- Between two hosts over the sim transport, with the fixed-key Noise + value and yamux + (PRINCIPLES 1.4): + - a request through the host gets its chunks, with whole and with + split writes; + - an id the peer does not serve fails the open, since + multistream-select answers `na`; + - a silent peer fails the request at its deadline; + - a connection closed under a parked read ends the request. - The error/EOF distinction: a framing failure returns the protocol error, and a clean stream end returns `finished`, through different channels (§5.2). An end inside a chunk is an early EOF.