-- | Identify protocol implementation (specs/identify).
--
-- Protocol ID: /ipfs/id/1.0.0
--
-- After a connection is established, both sides exchange IdentifyInfo
-- messages to learn about each other's capabilities, listen addresses,
-- and agent version. Like all libp2p protobuf streams, the message is
-- varint-length-delimited on the wire: uvarint(len) ++ protobuf. This
-- matches the delimited reader/writer used by go-libp2p (pbio),
-- rust-libp2p, and js-libp2p.
--
-- Also implements Identify Push (/ipfs/id/push/1.0.0) for proactive
-- updates when local state changes.
module LibP2P.Protocol.Identify
  ( -- * Protocol IDs
    identifyProtocolId
  , identifyPushProtocolId
    -- * Protocol logic
  , handleIdentify
  , requestIdentify
  , handleIdentifyPush
  , pushIdentify
  , mergeIdentify
    -- * Identify on connect
  , identifyPeer
  , identifyTimeoutMicros
    -- * Building local info
  , buildLocalIdentify
    -- * Registration
  , registerIdentifyHandlers
    -- * Wire framing
  , encodeFramedIdentify
  , readFramedIdentify
  ) where

import Control.Applicative ((<|>))
import Control.Concurrent.STM (atomically, modifyTVar', readTVar, writeTVar)
import Control.Exception (SomeException, bracket, catch, finally, try)
import Control.Monad (void)
import System.Timeout (timeout)
import qualified Data.ByteString as BS
import qualified Data.Map.Strict as Map
import LibP2P.Core.Varint (decodeUvarint, encodeUvarint)
import LibP2P.Crypto.PeerId (PeerId, fromPublicKey, peerIdBytes)
import LibP2P.Crypto.PeerRecord
  ( PeerRecord (..)
  , sealPeerRecord
  , timestampSeq
  )
import LibP2P.Switch.CertifiedRecords
  ( CertifiedRecord (..)
  , consumeCertifiedRecord
  , verifyPeerRecord
  )
import LibP2P.Crypto.Protobuf (decodePublicKey, encodePublicKey)
import LibP2P.Crypto.Key (kpPublic)
import LibP2P.Crypto.SignedEnvelope (encodeSignedEnvelope)
import LibP2P.Multiaddr.Codec (encodeProtocols)
import LibP2P.Multiaddr (Multiaddr (..))
import LibP2P.MultistreamSelect.Negotiation
  ( ProtocolId
  , StreamIO (..)
  , closeQuietly
  , negotiateInitiator
  , NegotiationResult (..)
  , readExactBounded
  )
import LibP2P.Protocol.Identify.Message
  ( IdentifyInfo (..)
  , decodeIdentify
  , encodeIdentify
  , maxIdentifySize
  )
import LibP2P.Switch.ConnPool (allConns)
import LibP2P.Switch.Types
  ( ActiveListener (..)
  , Connection (..)
  , MuxerSession (..)
  , Switch (..)
  )

-- | Identify protocol ID.
identifyProtocolId :: ProtocolId
identifyProtocolId :: Text
identifyProtocolId = Text
"/ipfs/id/1.0.0"

-- | Identify Push protocol ID.
identifyPushProtocolId :: ProtocolId
identifyPushProtocolId :: Text
identifyPushProtocolId = Text
"/ipfs/id/push/1.0.0"

-- | Handle an inbound Identify request (responder side).
--
-- Sends our local IdentifyInfo as a varint-length-prefixed protobuf,
-- then closes the stream (per specs/identify: respond and close).
-- The connection provides the remote address used for observedAddr.
handleIdentify :: Switch -> Connection -> StreamIO -> IO ()
handleIdentify :: Switch -> Connection -> StreamIO -> IO ()
handleIdentify Switch
sw Connection
conn StreamIO
stream =
  (do
    info <- Switch -> Maybe Connection -> IO IdentifyInfo
buildLocalIdentify Switch
sw (Connection -> Maybe Connection
forall a. a -> Maybe a
Just Connection
conn)
    streamWrite stream (encodeFramedIdentify info))
    IO () -> IO () -> IO ()
forall a b. IO a -> IO b -> IO a
`finally` StreamIO -> IO ()
closeQuietly StreamIO
stream

-- | Timeout for one Identify exchange: 5 seconds.
--
-- Matches go-libp2p's @identify.DefaultTimeout@, which it applies to all
-- id interactions in both directions.
identifyTimeoutMicros :: Int
identifyTimeoutMicros :: Int
identifyTimeoutMicros = Int
5000000

-- | Request Identify from a remote peer (initiator side).
--
-- Opens a new stream, negotiates /ipfs/id/1.0.0, then reads one
-- varint-length-prefixed protobuf message. The publicKey field is
-- validated against the connection's authenticated peer id (see
-- 'validatePublicKey').
--
-- Identify is a one-shot exchange, so the stream is closed on every exit
-- path — success, negotiation refusal, decode failure, timeout and
-- exception alike. A caller that runs this per connection would
-- otherwise leak a half-open stream per connection. The whole exchange
-- is bounded by 'identifyTimeoutMicros': a peer that negotiates and then
-- never answers must not pin a stream and a thread forever.
requestIdentify :: Connection -> IO (Either String IdentifyInfo)
requestIdentify :: Connection -> IO (Either [Char] IdentifyInfo)
requestIdentify Connection
conn = do
  outcome <- IO (Maybe (Either [Char] IdentifyInfo))
-> IO (Either SomeException (Maybe (Either [Char] IdentifyInfo)))
forall e a. Exception e => IO a -> IO (Either e a)
try (IO (Maybe (Either [Char] IdentifyInfo))
 -> IO (Either SomeException (Maybe (Either [Char] IdentifyInfo))))
-> IO (Maybe (Either [Char] IdentifyInfo))
-> IO (Either SomeException (Maybe (Either [Char] IdentifyInfo)))
forall a b. (a -> b) -> a -> b
$ IO StreamIO
-> (StreamIO -> IO ())
-> (StreamIO -> IO (Maybe (Either [Char] IdentifyInfo)))
-> IO (Maybe (Either [Char] IdentifyInfo))
forall a b c. IO a -> (a -> IO b) -> (a -> IO c) -> IO c
bracket (MuxerSession -> IO StreamIO
muxOpenStream (Connection -> MuxerSession
connSession Connection
conn)) StreamIO -> IO ()
closeQuietly StreamIO -> IO (Maybe (Either [Char] IdentifyInfo))
exchange
  pure $ case outcome of
    Left (SomeException
e :: SomeException) -> [Char] -> Either [Char] IdentifyInfo
forall a b. a -> Either a b
Left ([Char]
"identify failed: " [Char] -> [Char] -> [Char]
forall a. [a] -> [a] -> [a]
++ SomeException -> [Char]
forall a. Show a => a -> [Char]
show SomeException
e)
    Right Maybe (Either [Char] IdentifyInfo)
Nothing             -> [Char] -> Either [Char] IdentifyInfo
forall a b. a -> Either a b
Left [Char]
"identify timed out"
    Right (Just Either [Char] IdentifyInfo
result)       -> Either [Char] IdentifyInfo
result
  where
    exchange :: StreamIO -> IO (Maybe (Either [Char] IdentifyInfo))
exchange StreamIO
stream = Int
-> IO (Either [Char] IdentifyInfo)
-> IO (Maybe (Either [Char] IdentifyInfo))
forall a. Int -> IO a -> IO (Maybe a)
timeout Int
identifyTimeoutMicros (IO (Either [Char] IdentifyInfo)
 -> IO (Maybe (Either [Char] IdentifyInfo)))
-> IO (Either [Char] IdentifyInfo)
-> IO (Maybe (Either [Char] IdentifyInfo))
forall a b. (a -> b) -> a -> b
$ do
      negotiated <- StreamIO -> [Text] -> IO NegotiationResult
negotiateInitiator StreamIO
stream [Text
identifyProtocolId]
      case negotiated of
        NegotiationResult
NoProtocol -> Either [Char] IdentifyInfo -> IO (Either [Char] IdentifyInfo)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ([Char] -> Either [Char] IdentifyInfo
forall a b. a -> Either a b
Left [Char]
"remote does not support identify")
        Accepted Text
_ ->
          (IdentifyInfo -> IdentifyInfo)
-> Either [Char] IdentifyInfo -> Either [Char] IdentifyInfo
forall a b. (a -> b) -> Either [Char] a -> Either [Char] b
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
fmap (PeerId -> IdentifyInfo -> IdentifyInfo
validateIdentify (Connection -> PeerId
connPeerId Connection
conn))
            (Either [Char] IdentifyInfo -> Either [Char] IdentifyInfo)
-> IO (Either [Char] IdentifyInfo)
-> IO (Either [Char] IdentifyInfo)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> StreamIO -> Int -> IO (Either [Char] IdentifyInfo)
readFramedIdentify StreamIO
stream Int
maxIdentifySize

-- | Run Identify against a freshly established connection and record the
-- result in the peer store (specs/identify).
--
-- This is what makes a peer's advertised addresses and protocols known
-- to us: without it 'swPeerStore' only ever fills from an inbound push,
-- so nothing is known about a peer we dialled or accepted. go-libp2p
-- drives its IDService from the swarm's Connected notification for the
-- same reason.
--
-- Failure is returned rather than thrown: the connection stays usable,
-- and no peer store entry is created. There is no retry — a peer that
-- does not answer Identify now will be picked up by a later push, if it
-- sends one.
identifyPeer :: Switch -> Connection -> IO (Either String ())
identifyPeer :: Switch -> Connection -> IO (Either [Char] ())
identifyPeer Switch
sw Connection
conn = do
  result <- Connection -> IO (Either [Char] IdentifyInfo)
requestIdentify Connection
conn
  case result of
    Left [Char]
err -> Either [Char] () -> IO (Either [Char] ())
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ([Char] -> Either [Char] ()
forall a b. a -> Either a b
Left [Char]
err)
    Right IdentifyInfo
info -> do
      Switch -> PeerId -> IdentifyInfo -> IO ()
storeIdentify Switch
sw (Connection -> PeerId
connPeerId Connection
conn) IdentifyInfo
info
      Either [Char] () -> IO (Either [Char] ())
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (() -> Either [Char] ()
forall a b. b -> Either a b
Right ())

-- | Handle an inbound Identify Push (responder side).
--
-- Reads the pushed varint-length-prefixed IdentifyInfo from the remote
-- peer. The length prefix is the message boundary — identify push has
-- no stream-close boundary to fall back on. The local stream side is
-- still closed on every exit path so a one-shot push cannot leak a
-- half-open Yamux stream (go-libp2p's handleIdentifyResponse does the
-- same with defer s.Close()).
--
-- The pushed info is merged into the existing peer entry via
-- 'mergeIdentify': pushes may be partial updates, so fields absent
-- from the message must not erase what we already know.
handleIdentifyPush :: Switch -> Connection -> StreamIO -> IO ()
handleIdentifyPush :: Switch -> Connection -> StreamIO -> IO ()
handleIdentifyPush Switch
sw Connection
conn StreamIO
stream =
  (do
    infoOrErr <- StreamIO -> Int -> IO (Either [Char] IdentifyInfo)
readFramedIdentify StreamIO
stream Int
maxIdentifySize
    case infoOrErr of
      Left [Char]
_ -> () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
      Right IdentifyInfo
rawInfo ->
        Switch -> PeerId -> IdentifyInfo -> IO ()
storeIdentify Switch
sw (Connection -> PeerId
connPeerId Connection
conn)
          (PeerId -> IdentifyInfo -> IdentifyInfo
validateIdentify (Connection -> PeerId
connPeerId Connection
conn) IdentifyInfo
rawInfo))
    IO () -> IO () -> IO ()
forall a b. IO a -> IO b -> IO a
`finally` StreamIO -> IO ()
closeQuietly StreamIO
stream

-- | Merge validated Identify info into the peer store, enforcing RFC
-- 0003 record freshness on the way in.
--
-- Shared by the push responder and by 'identifyPeer' so both follow the
-- same rule: an update is merged into what is already known via
-- 'mergeIdentify' rather than replacing it, because a push may be a
-- partial update and must not erase fields it omits.
--
-- A signed peer record that is not strictly newer than the one already
-- retained for this peer is refused (see 'consumeCertifiedRecord'). Its
-- addresses have already been applied to 'idListenAddrs' by
-- 'validateSignedPeerRecord', so a refused record has to have that
-- undone: both the address list and the envelope are cleared from the
-- update, which leaves 'mergeIdentify' holding on to the certified
-- addresses and envelope already known. A replay therefore changes
-- nothing, which is the point of the rule.
--
-- The envelope is opened out here rather than inside the transaction so
-- an STM retry cannot make us re-verify a signature.
storeIdentify :: Switch -> PeerId -> IdentifyInfo -> IO ()
storeIdentify :: Switch -> PeerId -> IdentifyInfo -> IO ()
storeIdentify Switch
sw PeerId
peerId IdentifyInfo
info = do
  let offered :: Maybe CertifiedRecord
offered = do
        envBytes <- IdentifyInfo -> Maybe ByteString
idSignedPeerRecord IdentifyInfo
info
        either (const Nothing) Just (verifyPeerRecord peerId envBytes)
  STM () -> IO ()
forall a. STM a -> IO a
atomically (STM () -> IO ()) -> STM () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
    fresh <- case Maybe CertifiedRecord
offered of
      Maybe CertifiedRecord
Nothing     -> Bool -> STM Bool
forall a. a -> STM a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Bool
True
      Just CertifiedRecord
record -> TVar (Map PeerId CertifiedRecord)
-> PeerId -> CertifiedRecord -> STM Bool
consumeCertifiedRecord (Switch -> TVar (Map PeerId CertifiedRecord)
swCertifiedRecords Switch
sw) PeerId
peerId CertifiedRecord
record
    let update
          | Bool
fresh = IdentifyInfo
info
          | Bool
otherwise = IdentifyInfo
info { idSignedPeerRecord = Nothing, idListenAddrs = [] }
    store <- readTVar (swPeerStore sw)
    let merged = IdentifyInfo
-> (IdentifyInfo -> IdentifyInfo)
-> Maybe IdentifyInfo
-> IdentifyInfo
forall b a. b -> (a -> b) -> Maybe a -> b
maybe IdentifyInfo
update (IdentifyInfo -> IdentifyInfo -> IdentifyInfo
`mergeIdentify` IdentifyInfo
update) (PeerId -> Map PeerId IdentifyInfo -> Maybe IdentifyInfo
forall k a. Ord k => k -> Map k a -> Maybe a
Map.lookup PeerId
peerId Map PeerId IdentifyInfo
store)
    writeTVar (swPeerStore sw) (Map.insert peerId merged store)

-- | Validate the identity-bound fields of a received Identify message
-- against the peer id authenticated by the security handshake.
validateIdentify :: PeerId -> IdentifyInfo -> IdentifyInfo
validateIdentify :: PeerId -> IdentifyInfo -> IdentifyInfo
validateIdentify PeerId
remotePeer =
  PeerId -> IdentifyInfo -> IdentifyInfo
validateSignedPeerRecord PeerId
remotePeer (IdentifyInfo -> IdentifyInfo)
-> (IdentifyInfo -> IdentifyInfo) -> IdentifyInfo -> IdentifyInfo
forall b c a. (b -> c) -> (a -> b) -> a -> c
. PeerId -> IdentifyInfo -> IdentifyInfo
validatePublicKey PeerId
remotePeer

-- | Enforce the identify spec's key/peer-id binding: the publicKey
-- field must derive the sender's peer id, which the security handshake
-- has already authenticated.
--
-- A key that fails to decode or derives a different peer id is an
-- identity claim the sender cannot back up, so it is dropped from the
-- message (matching go-libp2p, which discards the key and keeps the
-- connection — it is already authenticated). The rest of the message
-- is untouched, and previously known good data stays intact because
-- 'mergeIdentify' keeps the known key when the update carries none.
validatePublicKey :: PeerId -> IdentifyInfo -> IdentifyInfo
validatePublicKey :: PeerId -> IdentifyInfo -> IdentifyInfo
validatePublicKey PeerId
remotePeer IdentifyInfo
info = case IdentifyInfo -> Maybe ByteString
idPublicKey IdentifyInfo
info of
  Maybe ByteString
Nothing -> IdentifyInfo
info
  Just ByteString
keyBytes -> case ByteString -> Either [Char] PublicKey
decodePublicKey ByteString
keyBytes of
    Right PublicKey
pk | PublicKey -> PeerId
fromPublicKey PublicKey
pk PeerId -> PeerId -> Bool
forall a. Eq a => a -> a -> Bool
== PeerId
remotePeer -> IdentifyInfo
info
    Either [Char] PublicKey
_ -> IdentifyInfo
info { idPublicKey = Nothing }

-- | Verify a received signedPeerRecord (RFC 0003) against the
-- authenticated peer id: the envelope must open (valid signature,
-- payload type, key/record binding) and its signing key must derive
-- the peer id the security handshake authenticated.
--
-- A verified record's addresses are authoritative and replace the
-- unsigned listenAddrs (go-libp2p's certified addr book takes signed
-- addresses over unsigned ones). A record that fails verification is
-- dropped, keeping the unsigned listenAddrs as the fallback for peers
-- whose record we cannot trust.
--
-- Verification alone does not make a record current: a correctly signed
-- older record can be replayed. The sequence-number check that decides
-- whether it may replace what is already retained needs the peer store,
-- so it lives in 'storeIdentify'.
validateSignedPeerRecord :: PeerId -> IdentifyInfo -> IdentifyInfo
validateSignedPeerRecord :: PeerId -> IdentifyInfo -> IdentifyInfo
validateSignedPeerRecord PeerId
remotePeer IdentifyInfo
info = case IdentifyInfo -> Maybe ByteString
idSignedPeerRecord IdentifyInfo
info of
  Maybe ByteString
Nothing -> IdentifyInfo
info
  Just ByteString
envBytes -> case PeerId -> ByteString -> Either [Char] CertifiedRecord
verifyPeerRecord PeerId
remotePeer ByteString
envBytes of
    Right CertifiedRecord
record -> IdentifyInfo
info { idListenAddrs = crAddresses record }
    Left [Char]
_       -> IdentifyInfo
info { idSignedPeerRecord = Nothing }

-- | Merge a received (possibly partial) Identify update into the
-- previously known info for a peer.
--
-- Per specs/identify: "missing fields should be ignored, as peers may
-- choose to send partial updates containing only the fields whose
-- values have changed." Optional fields keep the known value when the
-- update omits them; repeated fields (protobuf cannot distinguish
-- absent from empty) keep the known list when the update's is empty
-- and are replaced wholesale otherwise, matching go-libp2p.
mergeIdentify :: IdentifyInfo -> IdentifyInfo -> IdentifyInfo
mergeIdentify :: IdentifyInfo -> IdentifyInfo -> IdentifyInfo
mergeIdentify IdentifyInfo
known IdentifyInfo
update = IdentifyInfo
  { idProtocolVersion :: Maybe Text
idProtocolVersion = IdentifyInfo -> Maybe Text
idProtocolVersion IdentifyInfo
update Maybe Text -> Maybe Text -> Maybe Text
forall a. Maybe a -> Maybe a -> Maybe a
forall (f :: * -> *) a. Alternative f => f a -> f a -> f a
<|> IdentifyInfo -> Maybe Text
idProtocolVersion IdentifyInfo
known
  , idAgentVersion :: Maybe Text
idAgentVersion    = IdentifyInfo -> Maybe Text
idAgentVersion IdentifyInfo
update Maybe Text -> Maybe Text -> Maybe Text
forall a. Maybe a -> Maybe a -> Maybe a
forall (f :: * -> *) a. Alternative f => f a -> f a -> f a
<|> IdentifyInfo -> Maybe Text
idAgentVersion IdentifyInfo
known
  , idPublicKey :: Maybe ByteString
idPublicKey       = IdentifyInfo -> Maybe ByteString
idPublicKey IdentifyInfo
update Maybe ByteString -> Maybe ByteString -> Maybe ByteString
forall a. Maybe a -> Maybe a -> Maybe a
forall (f :: * -> *) a. Alternative f => f a -> f a -> f a
<|> IdentifyInfo -> Maybe ByteString
idPublicKey IdentifyInfo
known
  , idListenAddrs :: [ByteString]
idListenAddrs     = [ByteString] -> [ByteString] -> [ByteString]
forall a. [a] -> [a] -> [a]
replaceUnlessEmpty (IdentifyInfo -> [ByteString]
idListenAddrs IdentifyInfo
known) (IdentifyInfo -> [ByteString]
idListenAddrs IdentifyInfo
update)
  , idObservedAddr :: Maybe ByteString
idObservedAddr    = IdentifyInfo -> Maybe ByteString
idObservedAddr IdentifyInfo
update Maybe ByteString -> Maybe ByteString -> Maybe ByteString
forall a. Maybe a -> Maybe a -> Maybe a
forall (f :: * -> *) a. Alternative f => f a -> f a -> f a
<|> IdentifyInfo -> Maybe ByteString
idObservedAddr IdentifyInfo
known
  , idProtocols :: [Text]
idProtocols       = [Text] -> [Text] -> [Text]
forall a. [a] -> [a] -> [a]
replaceUnlessEmpty (IdentifyInfo -> [Text]
idProtocols IdentifyInfo
known) (IdentifyInfo -> [Text]
idProtocols IdentifyInfo
update)
  , idSignedPeerRecord :: Maybe ByteString
idSignedPeerRecord = IdentifyInfo -> Maybe ByteString
idSignedPeerRecord IdentifyInfo
update Maybe ByteString -> Maybe ByteString -> Maybe ByteString
forall a. Maybe a -> Maybe a -> Maybe a
forall (f :: * -> *) a. Alternative f => f a -> f a -> f a
<|> IdentifyInfo -> Maybe ByteString
idSignedPeerRecord IdentifyInfo
known
  }
  where
    replaceUnlessEmpty :: [a] -> [a] -> [a]
replaceUnlessEmpty [a]
old [] = [a]
old
    replaceUnlessEmpty [a]
_ [a]
new  = [a]
new

-- | Push our current IdentifyInfo to every connected peer (sender side
-- of /ipfs/id/push/1.0.0).
--
-- Per specs/identify: open a stream to each remote peer, negotiate the
-- push protocol id, send one Identify message and close the stream.
-- Call this whenever local state advertised via identify changes
-- (listen addresses, registered protocols). Failures on individual
-- peers (e.g. push protocol not supported) are ignored.
pushIdentify :: Switch -> IO ()
pushIdentify :: Switch -> IO ()
pushIdentify Switch
sw = do
  conns <- STM [Connection] -> IO [Connection]
forall a. STM a -> IO a
atomically (STM [Connection] -> IO [Connection])
-> STM [Connection] -> IO [Connection]
forall a b. (a -> b) -> a -> b
$ TVar (Map PeerId [Connection]) -> STM [Connection]
allConns (Switch -> TVar (Map PeerId [Connection])
swConnPool Switch
sw)
  mapM_ (\Connection
conn -> Connection -> IO ()
pushToConn Connection
conn IO () -> (SomeException -> IO ()) -> IO ()
forall e a. Exception e => IO a -> (e -> IO a) -> IO a
`catch` \(SomeException
_ :: SomeException) -> () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()) conns
  where
    pushToConn :: Connection -> IO ()
pushToConn Connection
conn = do
      stream <- MuxerSession -> IO StreamIO
muxOpenStream (Connection -> MuxerSession
connSession Connection
conn)
      result <- negotiateInitiator stream [identifyPushProtocolId]
      case result of
        Accepted Text
_ -> do
          info <- Switch -> Maybe Connection -> IO IdentifyInfo
buildLocalIdentify Switch
sw (Connection -> Maybe Connection
forall a. a -> Maybe a
Just Connection
conn)
          streamWrite stream (encodeFramedIdentify info)
          streamClose stream
        NegotiationResult
NoProtocol -> StreamIO -> IO ()
streamClose StreamIO
stream

-- | Build our local IdentifyInfo from Switch state, including a signed
-- peer record (RFC 0003) over our listen addresses, sealed with the
-- identity key.
buildLocalIdentify :: Switch -> Maybe Connection -> IO IdentifyInfo
buildLocalIdentify :: Switch -> Maybe Connection -> IO IdentifyInfo
buildLocalIdentify Switch
sw Maybe Connection
mConn = do
  (protocols, listenAddrs) <- STM ([Text], [Multiaddr]) -> IO ([Text], [Multiaddr])
forall a. STM a -> IO a
atomically (STM ([Text], [Multiaddr]) -> IO ([Text], [Multiaddr]))
-> STM ([Text], [Multiaddr]) -> IO ([Text], [Multiaddr])
forall a b. (a -> b) -> a -> b
$ do
    protos <- Map Text (Connection -> StreamIO -> IO ()) -> [Text]
forall k a. Map k a -> [k]
Map.keys (Map Text (Connection -> StreamIO -> IO ()) -> [Text])
-> STM (Map Text (Connection -> StreamIO -> IO ())) -> STM [Text]
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> TVar (Map Text (Connection -> StreamIO -> IO ()))
-> STM (Map Text (Connection -> StreamIO -> IO ()))
forall a. TVar a -> STM a
readTVar (Switch -> TVar (Map Text (Connection -> StreamIO -> IO ()))
swProtocols Switch
sw)
    listeners <- readTVar (swListeners sw)
    pure (protos, map alAddress listeners)
  seqNo <- timestampSeq
  let addrBytes = (Multiaddr -> ByteString) -> [Multiaddr] -> [ByteString]
forall a b. (a -> b) -> [a] -> [b]
map (\(Multiaddr [Protocol]
ps) -> [Protocol] -> ByteString
encodeProtocols [Protocol]
ps) [Multiaddr]
listenAddrs
      record = PeerRecord
        { prPeerId :: ByteString
prPeerId    = PeerId -> ByteString
peerIdBytes (Switch -> PeerId
swLocalPeerId Switch
sw)
        , prSeq :: Word64
prSeq       = Word64
seqNo
        , prAddresses :: [ByteString]
prAddresses = [ByteString]
addrBytes
        }
      -- Sealing our own record with our own identity key cannot fail;
      -- if it somehow does, the optional field is omitted.
      signedRecord = ([Char] -> Maybe ByteString)
-> (SignedEnvelope -> Maybe ByteString)
-> Either [Char] SignedEnvelope
-> Maybe ByteString
forall a c b. (a -> c) -> (b -> c) -> Either a b -> c
either (Maybe ByteString -> [Char] -> Maybe ByteString
forall a b. a -> b -> a
const Maybe ByteString
forall a. Maybe a
Nothing) (ByteString -> Maybe ByteString
forall a. a -> Maybe a
Just (ByteString -> Maybe ByteString)
-> (SignedEnvelope -> ByteString)
-> SignedEnvelope
-> Maybe ByteString
forall b c a. (b -> c) -> (a -> b) -> a -> c
. SignedEnvelope -> ByteString
encodeSignedEnvelope)
                       (KeyPair -> PeerRecord -> Either [Char] SignedEnvelope
sealPeerRecord (Switch -> KeyPair
swIdentityKey Switch
sw) PeerRecord
record)
  pure IdentifyInfo
    { idProtocolVersion = Just "ipfs/0.1.0"
    , idAgentVersion    = Just "libp2p-hs/0.1.0"
    , idPublicKey       = Just (encodePublicKey (kpPublic (swIdentityKey sw)))
    , idListenAddrs     = addrBytes
    , idObservedAddr    = (\(Multiaddr [Protocol]
ps) -> [Protocol] -> ByteString
encodeProtocols [Protocol]
ps) . connRemoteAddr <$> mConn
    , idProtocols       = protocols
    , idSignedPeerRecord = signedRecord
    }

-- | Register Identify protocol handlers on the Switch.
--
-- Registers:
--   /ipfs/id/1.0.0      — respond to Identify requests
--   /ipfs/id/push/1.0.0 — handle Identify Push from remote
registerIdentifyHandlers :: Switch -> IO ()
registerIdentifyHandlers :: Switch -> IO ()
registerIdentifyHandlers Switch
sw = do
  STM () -> IO ()
forall a. STM a -> IO a
atomically (STM () -> IO ()) -> STM () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
    protos <- TVar (Map Text (Connection -> StreamIO -> IO ()))
-> STM (Map Text (Connection -> StreamIO -> IO ()))
forall a. TVar a -> STM a
readTVar (Switch -> TVar (Map Text (Connection -> StreamIO -> IO ()))
swProtocols Switch
sw)
    let protos' = Text
-> (Connection -> StreamIO -> IO ())
-> Map Text (Connection -> StreamIO -> IO ())
-> Map Text (Connection -> StreamIO -> IO ())
forall k a. Ord k => k -> a -> Map k a -> Map k a
Map.insert Text
identifyProtocolId (Switch -> Connection -> StreamIO -> IO ()
handleIdentify Switch
sw) Map Text (Connection -> StreamIO -> IO ())
protos
        protos'' = Text
-> (Connection -> StreamIO -> IO ())
-> Map Text (Connection -> StreamIO -> IO ())
-> Map Text (Connection -> StreamIO -> IO ())
forall k a. Ord k => k -> a -> Map k a -> Map k a
Map.insert Text
identifyPushProtocolId (Switch -> Connection -> StreamIO -> IO ()
handleIdentifyPush Switch
sw) Map Text (Connection -> StreamIO -> IO ())
protos'
    writeTVar (swProtocols sw) protos''
  -- Identify every connection as it comes up, inbound and outbound, so
  -- the peer store reflects peers we dialled and accepted rather than
  -- only those that pushed to us. The notifier discards the outcome;
  -- callers that need to observe failure use 'identifyPeer' directly.
  STM () -> IO ()
forall a. STM a -> IO a
atomically (STM () -> IO ()) -> STM () -> IO ()
forall a b. (a -> b) -> a -> b
$ TVar [Connection -> IO ()]
-> ([Connection -> IO ()] -> [Connection -> IO ()]) -> STM ()
forall a. TVar a -> (a -> a) -> STM ()
modifyTVar' (Switch -> TVar [Connection -> IO ()]
swNotifiers Switch
sw) (IO (Either [Char] ()) -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (IO (Either [Char] ()) -> IO ())
-> (Connection -> IO (Either [Char] ())) -> Connection -> IO ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Switch -> Connection -> IO (Either [Char] ())
identifyPeer Switch
sw (Connection -> IO ())
-> [Connection -> IO ()] -> [Connection -> IO ()]
forall a. a -> [a] -> [a]
:)

-- | Encode an IdentifyInfo with its uvarint length prefix, as written
-- on the wire: uvarint(len) ++ protobuf.
encodeFramedIdentify :: IdentifyInfo -> BS.ByteString
encodeFramedIdentify :: IdentifyInfo -> ByteString
encodeFramedIdentify IdentifyInfo
info =
  let payload :: ByteString
payload = IdentifyInfo -> ByteString
encodeIdentify IdentifyInfo
info
  in Word64 -> ByteString
encodeUvarint (Int -> Word64
forall a b. (Integral a, Num b) => a -> b
fromIntegral (ByteString -> Int
BS.length ByteString
payload)) ByteString -> ByteString -> ByteString
forall a. Semigroup a => a -> a -> a
<> ByteString
payload

-- | Read one varint-length-prefixed Identify message from a stream.
--
-- Reads the uvarint length prefix, then exactly that many payload
-- bytes, and decodes the protobuf. Rejects messages larger than
-- maxSize before reading the payload.
readFramedIdentify :: StreamIO -> Int -> IO (Either String IdentifyInfo)
readFramedIdentify :: StreamIO -> Int -> IO (Either [Char] IdentifyInfo)
readFramedIdentify StreamIO
stream Int
maxSize = IO (Either [Char] IdentifyInfo)
readFramed IO (Either [Char] IdentifyInfo)
-> (SomeException -> IO (Either [Char] IdentifyInfo))
-> IO (Either [Char] IdentifyInfo)
forall e a. Exception e => IO a -> (e -> IO a) -> IO a
`catch` SomeException -> IO (Either [Char] IdentifyInfo)
onError
  where
    onError :: SomeException -> IO (Either String IdentifyInfo)
    onError :: SomeException -> IO (Either [Char] IdentifyInfo)
onError SomeException
e = Either [Char] IdentifyInfo -> IO (Either [Char] IdentifyInfo)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ([Char] -> Either [Char] IdentifyInfo
forall a b. a -> Either a b
Left ([Char]
"identify stream read failed: " [Char] -> [Char] -> [Char]
forall a. [a] -> [a] -> [a]
++ SomeException -> [Char]
forall a. Show a => a -> [Char]
show SomeException
e))

    readFramed :: IO (Either [Char] IdentifyInfo)
readFramed = do
      varintBytes <- StreamIO -> IO ByteString
readVarintBytes StreamIO
stream
      case decodeUvarint varintBytes of
        Left [Char]
err -> Either [Char] IdentifyInfo -> IO (Either [Char] IdentifyInfo)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ([Char] -> Either [Char] IdentifyInfo
forall a b. a -> Either a b
Left ([Char]
"identify length prefix decode error: " [Char] -> [Char] -> [Char]
forall a. [a] -> [a] -> [a]
++ [Char]
err))
        Right (Word64
len, ByteString
_) -> do
          let msgLen :: Int
msgLen = Word64 -> Int
forall a b. (Integral a, Num b) => a -> b
fromIntegral Word64
len :: Int
          if Int
msgLen Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
> Int
maxSize
            then Either [Char] IdentifyInfo -> IO (Either [Char] IdentifyInfo)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ([Char] -> Either [Char] IdentifyInfo
forall a b. a -> Either a b
Left ([Char]
"identify message too large: "
                             [Char] -> [Char] -> [Char]
forall a. [a] -> [a] -> [a]
++ Int -> [Char]
forall a. Show a => a -> [Char]
show Int
msgLen [Char] -> [Char] -> [Char]
forall a. [a] -> [a] -> [a]
++ [Char]
" > " [Char] -> [Char] -> [Char]
forall a. [a] -> [a] -> [a]
++ Int -> [Char]
forall a. Show a => a -> [Char]
show Int
maxSize))
            else do
              payloadOrErr <- StreamIO -> Int -> Int -> IO (Either [Char] ByteString)
readExactBounded StreamIO
stream Int
maxSize Int
msgLen
              case payloadOrErr of
                Left [Char]
err -> Either [Char] IdentifyInfo -> IO (Either [Char] IdentifyInfo)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ([Char] -> Either [Char] IdentifyInfo
forall a b. a -> Either a b
Left ([Char]
"identify read error: " [Char] -> [Char] -> [Char]
forall a. [a] -> [a] -> [a]
++ [Char]
err))
                Right ByteString
payload -> case ByteString -> Either ParseError IdentifyInfo
decodeIdentify ByteString
payload of
                  Left ParseError
parseErr ->
                    Either [Char] IdentifyInfo -> IO (Either [Char] IdentifyInfo)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ([Char] -> Either [Char] IdentifyInfo
forall a b. a -> Either a b
Left ([Char]
"identify protobuf decode error: " [Char] -> [Char] -> [Char]
forall a. [a] -> [a] -> [a]
++ ParseError -> [Char]
forall a. Show a => a -> [Char]
show ParseError
parseErr))
                  Right IdentifyInfo
info -> Either [Char] IdentifyInfo -> IO (Either [Char] IdentifyInfo)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (IdentifyInfo -> Either [Char] IdentifyInfo
forall a b. b -> Either a b
Right IdentifyInfo
info)

-- | Read the bytes of one unsigned varint from a stream (up to 10 bytes).
readVarintBytes :: StreamIO -> IO BS.ByteString
readVarintBytes :: StreamIO -> IO ByteString
readVarintBytes StreamIO
stream = [Word8] -> Int -> IO ByteString
forall {t}. (Ord t, Num t) => [Word8] -> t -> IO ByteString
go [] (Int
0 :: Int)
  where
    go :: [Word8] -> t -> IO ByteString
go [Word8]
acc t
n
      | t
n t -> t -> Bool
forall a. Ord a => a -> a -> Bool
>= t
10 = ByteString -> IO ByteString
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ([Word8] -> ByteString
BS.pack ([Word8] -> [Word8]
forall a. [a] -> [a]
reverse [Word8]
acc))  -- max varint length
      | Bool
otherwise = do
          b <- StreamIO -> IO Word8
streamReadByte StreamIO
stream
          if b < 0x80
            then pure (BS.pack (reverse (b : acc)))
            else go (b : acc) (n + 1)