plexus-protocol 0.5.0.0 → 0.5.2.0

raw patch · 7 files changed

+684/−48 lines, 7 filesdep +crypton-x509-storedep +crypton-x509-systemdep +tlsdep ~networkPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

Dependencies added: crypton-x509-store, crypton-x509-system, tls

Dependency ranges changed: network

API changes (from Hackage documentation)

Files

• plexus-protocol.cabal +7/−2
• src/Plexus.hs +2/−0
• src/Plexus/Auth.hs +202/−0
• src/Plexus/Client.hs +284/−20
• src/Plexus/Schema/Recursive.hs +91/−9
• src/Plexus/Transport.hs +35/−7
• src/Plexus/Types.hs +63/−10

@@ -1,6 +1,6 @@ cabal-version:      3.0 name:               plexus-protocol-version:            0.5.0.0+version:            0.5.2.0 synopsis:           Plexus RPC protocol types and client license:            MIT author:@@ -14,6 +14,7 @@     Plexus.Transport     Plexus     Plexus.Types+    Plexus.Auth     Plexus.Schema     Plexus.Schema.Cache     Plexus.Schema.Recursive@@ -26,7 +27,11 @@     case-insensitive >= 1.2,     containers >= 0.6,     websockets >= 0.13,-    network >= 3.1,+    network >= 3.1.2,+    -- SCAFFOLD-21: wss — TLS, the certificate checked against the system store+    tls >= 2.1 && < 2.5,+    crypton-x509-store >= 1.6 && < 1.10,+    crypton-x509-system >= 1.6 && < 1.10,     async >= 2.2,     stm >= 2.5,     streaming >= 0.2,

@@ -34,6 +34,8 @@   ( -- * Connection (re-exported from Plexus.Client)     SubstrateConnection   , SubstrateConfig(..)+  , TlsOptions(..)+  , defaultTlsOptions   , defaultConfig   , connect   , disconnect

@@ -0,0 +1,202 @@+-- | AUTH-4 / AUTH-2 / AUTH-5: what a client parses of a hub's auth.+--+-- Every field is optional on the wire: a hub without a gate (every hub+-- before AUTH, and any hub that leaves login off) answers none of it, and+-- everything here parses its absence as 'Nothing' or empty.+--+-- * '_info' gains @view@ (@anonymous@ | @caller@), the view's @hash@, and,+--   to an anonymous caller, an @auth@ block naming the methods it may call+--   and the RFC 9728 resource metadata URL ('BackendInfo').+-- * A refusal is an error item with @error_kind: "forbidden"@ naming the+--   grant it needed, the scope it was asked on, the roles that list the+--   grant and where the caller holds it ('Refusal').+module Plexus.Auth+  ( -- * The view (AUTH-4)+    View(..)+  , viewText+  , BackendInfo(..)+  , AuthInfo(..)+  , parseBackendInfo++    -- * Refusals (AUTH-2)+  , Refusal(..)+  , parseRefusal+  , refusalFromObject+  ) where++import Control.Applicative ((<|>))+import Data.Aeson+import qualified Data.Aeson.KeyMap as KM+import Data.Aeson.Types (Parser, parseMaybe)+import Data.Text (Text)+import qualified Data.Text as T+import GHC.Generics (Generic)++-- ============================================================================+-- The view+-- ============================================================================++-- | Which part of a hub the caller sees: an anonymous caller only how to+-- sign in; a signed-in caller what it can use.+data View = Anonymous | Caller+  deriving stock (Show, Eq, Ord, Generic)++viewText :: View -> Text+viewText Anonymous = "anonymous"+viewText Caller = "caller"++-- | @"anonymous"@ / @"caller"@, or an object naming it as @kind@ (a+-- tolerance: a hub may carry the view's hash beside it).+instance FromJSON View where+  parseJSON = \case+    String s -> fromName s+    Object o -> (o .: "kind" <|> o .: "view") >>= fromName+    v -> fail ("View: expected a string, got " <> show v)+    where+      fromName s = case T.toLower s of+        "anonymous" -> pure Anonymous+        "caller" -> pure Caller+        other -> fail ("View: unknown view " <> T.unpack other)++instance ToJSON View where+  toJSON = String . viewText++-- | The anonymous view's door: the methods it may call and where an HTTP+-- client reads how to get a token.+data AuthInfo = AuthInfo+  { authMethods          :: [Text]      -- ^ e.g. @["auth.providers","auth.login","auth.signup"]@+  , authResourceMetadata :: Maybe Text  -- ^ RFC 9728 @/.well-known/oauth-protected-resource@+  }+  deriving stock (Show, Eq, Generic)++instance FromJSON AuthInfo where+  parseJSON = withObject "AuthInfo" $ \o -> AuthInfo+    <$> o .:? "methods" .!= []+    <*> o .:? "resource_metadata"++instance ToJSON AuthInfo where+  toJSON AuthInfo{..} = object $+    ["methods" .= authMethods]+    <> maybe [] (\u -> ["resource_metadata" .= u]) authResourceMetadata++-- | What @_info@ answers. Only @backend@ is required.+data BackendInfo = BackendInfo+  { infoBackend  :: Text+  , infoView     :: Maybe View   -- ^ Nothing: a hub with no gate (today's)+  , infoViewHash :: Maybe Text   -- ^ the view's own hash (@view_hash@, else @hash@, else the view object's @hash@)+  , infoAuth     :: Maybe AuthInfo+  }+  deriving stock (Show, Eq, Generic)++instance FromJSON BackendInfo where+  parseJSON = withObject "BackendInfo" $ \o -> do+    backend <- o .: "backend"+    -- a view that does not parse is no view, never a failed _info+    view <- optionalLenient (o .:? "view")+    viewObjHash <- case KM.lookup "view" o of+      Just (Object vo) -> optionalLenient (vo .:? "hash")+      _ -> pure Nothing+    vhash <- optionalLenient (o .:? "view_hash")+    hash <- optionalLenient (o .:? "hash")+    auth <- optionalLenient (o .:? "auth")+    pure BackendInfo+      { infoBackend = backend+      , infoView = view+      , infoViewHash = vhash <|> viewObjHash <|> hash+      , infoAuth = auth+      }++instance ToJSON BackendInfo where+  toJSON BackendInfo{..} = object $+    ["backend" .= infoBackend]+    <> maybe [] (\v -> ["view" .= v]) infoView+    <> maybe [] (\h -> ["view_hash" .= h]) infoViewHash+    <> maybe [] (\a -> ["auth" .= a]) infoAuth++-- | Parse @_info@'s content.+parseBackendInfo :: Value -> Maybe BackendInfo+parseBackendInfo = parseMaybe parseJSON++-- | A field that may be present but malformed: Nothing rather than failing.+optionalLenient :: Parser (Maybe a) -> Parser (Maybe a)+optionalLenient p = p <|> pure Nothing++-- ============================================================================+-- Refusals+-- ============================================================================++-- | A typed refusal: @error_kind: "forbidden"@. Every field but the+-- message is optional; lists absent are empty.+data Refusal = Refusal+  { refusalGrant   :: Maybe Text   -- ^ @tickets.set_status@+  , refusalScope   :: Maybe Text   -- ^ where it was asked: @IDENTITY@+  , refusalRoles   :: [Text]       -- ^ the roles that list the grant+  , refusalHeldOn  :: [Text]       -- ^ where the caller holds it+  , refusalCeiling :: [Text]       -- ^ the session's ceiling, when the hub says it+  , refusalMessage :: Text+  }+  deriving stock (Show, Eq, Generic)++instance FromJSON Refusal where+  parseJSON = withObject "Refusal" $ \o -> do+    kind <- o .:? "error_kind"+    case kind :: Maybe Text of+      Just k | T.toLower k == "forbidden" -> refusalFields o+      _ -> fail "Refusal: error_kind is not forbidden"++instance ToJSON Refusal where+  toJSON Refusal{..} = object $+    [ "error_kind" .= ("forbidden" :: Text)+    , "message" .= refusalMessage+    , "roles" .= refusalRoles+    , "held_on" .= refusalHeldOn+    ]+    <> maybe [] (\g -> ["grant" .= g]) refusalGrant+    <> maybe [] (\s -> ["scope" .= s]) refusalScope+    <> (if null refusalCeiling then [] else ["ceiling" .= refusalCeiling])++refusalFields :: Object -> Parser Refusal+refusalFields o = do+  grant <- optionalLenient (o .:? "grant")+  scope <- optionalLenient (o .:? "scope")+  roles <- textList o "roles"+  held <- textList o "held_on"+  ceiling <- textList o "ceiling"+  msg <- optionalLenient (o .:? "message")+  pure Refusal+    { refusalGrant = grant+    , refusalScope = scope+    , refusalRoles = roles+    , refusalHeldOn = held+    , refusalCeiling = ceiling+    , refusalMessage = maybe (defaultMessage grant scope) id msg+    }+  where+    defaultMessage g s = case (g, s) of+      (Just g', Just s') -> "needs " <> g' <> " on " <> s'+      (Just g', Nothing) -> "needs " <> g'+      _ -> "forbidden"++-- | A list of strings, a single string, or a list whose elements are+-- objects naming a @scope@ / @role@ / @name@ — anything else is empty.+textList :: Object -> Key -> Parser [Text]+textList o k = pure $ case KM.lookup k o of+  Just (Array xs) -> concatMap one (foldr (:) [] xs)+  Just (String s) -> [s]+  _ -> []+  where+    one = \case+      String s -> [s]+      Object e+        | Just (String s) <- KM.lookup "scope" e -> [s]+        | Just (String s) <- KM.lookup "role" e -> [s]+        | Just (String s) <- KM.lookup "name" e -> [s]+      _ -> []++-- | A refusal from an error item's (or an in-band error event's) fields.+refusalFromObject :: Object -> Maybe Refusal+refusalFromObject = parseMaybe parseJSON . Object++-- | A refusal from any JSON value: an object with @error_kind: forbidden@.+parseRefusal :: Value -> Maybe Refusal+parseRefusal = parseMaybe parseJSON

@@ -12,22 +12,37 @@     SubstrateConnection   , connect   , disconnect+  , connectionLost +    -- * A lost connection (SYNAPSE-20)+  , LostConnection(..)+  , Keepalive(..)+  , keepaliveFromEnv+     -- * Core RPC primitive   , substrateRpc+  , substrateRpcWith      -- * Configuration   , SubstrateConfig(..)+  , TlsOptions(..)+  , defaultTlsOptions   , defaultConfig   , cookieHeader+  , hostHeader   ) where -import Control.Concurrent (forkIO)+import Control.Concurrent (forkIO, killThread, threadDelay)+import Control.Concurrent.MVar (MVar, modifyMVar, newMVar)+import System.IO.Unsafe (unsafePerformIO) import Control.Concurrent.Async (Async, async, cancel, wait) import Control.Concurrent.STM-import Control.Exception (SomeException, catch)-import Control.Monad (forever, void)+import Control.Exception (Exception, SomeException, bracketOnError, catch, fromException, throwIO, finally, try)+import Control.Monad (forever, void, when, unless)+import GHC.Clock (getMonotonicTime)+import System.Environment (lookupEnv) import System.Timeout (timeout)+import Text.Read (readMaybe) import Data.Aeson import qualified Data.Aeson.KeyMap as KM import qualified Data.ByteString.Lazy.Char8 as LBS@@ -38,8 +53,15 @@ import qualified Data.Text as T import Network.WebSockets (Connection) import qualified Network.WebSockets as WS+import qualified Network.WebSockets.Stream as WSS+import qualified Data.ByteString as BS import qualified Data.ByteString.Char8 as BS8 import qualified Data.CaseInsensitive as CI+import Data.X509.CertificateStore (CertificateStore, makeCertificateStore)+import Data.X509.File (readSignedObject)+import qualified Network.Socket as S+import qualified Network.TLS as TLS+import System.X509 (getSystemCertificateStore) import Streaming import qualified Streaming.Prelude as Str @@ -52,9 +74,29 @@   , substratePath    :: String   , substrateBackend :: Text    -- ^ Backend name (e.g., "plexus")   , substrateHeaders :: WS.Headers  -- ^ Extra HTTP upgrade headers (e.g., Cookie)+  , substrateTls     :: Maybe TlsOptions+    -- ^ SCAFFOLD-21: 'Nothing' dials @ws://@; 'Just' dials @wss://@ — TLS+    -- with the server's certificate verified against the system's trust+    -- store (plus 'tlsCaFiles') and its name ('substrateHost'), sent as SNI.   }   deriving stock (Show, Eq, Ord) +-- | SCAFFOLD-21: how a @wss://@ connection is made. The certificate is+-- always verified — the chain against the system's trust store (the+-- distribution's bundle; @SYSTEM_CERTIFICATE_PATH@ overrides where it is+-- read from) and every PEM file in 'tlsCaFiles' (a local CA, for a test+-- behind a proxy of one's own), the name against 'substrateHost'. There is+-- no switch that skips the check.+data TlsOptions = TlsOptions+  { tlsCaFiles :: [FilePath]+    -- ^ PEM files of certificates trusted beside the system's+  }+  deriving stock (Show, Eq, Ord)++-- | Verify against the system's trust store alone.+defaultTlsOptions :: TlsOptions+defaultTlsOptions = TlsOptions { tlsCaFiles = [] }+ -- | Default configuration for local development (requires backend) defaultConfig :: Text -> SubstrateConfig defaultConfig backend = SubstrateConfig@@ -63,8 +105,20 @@   , substratePath    = "/"   , substrateBackend = backend   , substrateHeaders = []+  , substrateTls     = Nothing   } +-- | SCAFFOLD-21: the upgrade's @Host@ — the host alone on the scheme's+-- default port (443 for wss, 80 for ws), else @host:port@.+hostHeader :: SubstrateConfig -> String+hostHeader SubstrateConfig{..}+  | substratePort == defaultPort = substrateHost+  | otherwise = bracketed <> ":" <> show substratePort+  where+    defaultPort = maybe 80 (const 443) substrateTls+    bracketed | ':' `elem` substrateHost = "[" <> substrateHost <> "]"  -- an IPv6 literal+              | otherwise = substrateHost+ -- | Build a Cookie header from a JWT token cookieHeader :: Text -> WS.Headers cookieHeader tok = [(CI.mk "Cookie", "access_token=" <> BS8.pack (T.unpack tok))]@@ -82,8 +136,66 @@   , scSubscriptions :: TVar (Map SubscriptionId (TQueue Value))   , scPendingReqs   :: TVar (Map RequestId PendingRequest)   , scReaderThread  :: Async ()+  , scClosed        :: TMVar Text+    -- ^ SYNAPSE-20: filled, once, with why the connection is gone — the+    -- reader ended (a close, a reset) or a keepalive ping went unanswered.+    -- Every call waiting on this connection waits on it too, so none is+    -- left waiting on a dead socket.+  , scKeepalive     :: Maybe (Async ())   } +-- | SYNAPSE-20: a call's connection was lost. 'lostSent' says whether the+-- call's request had gone out before it was.+data LostConnection = LostConnection+  { lostReason :: Text+  , lostSent   :: Bool+  }+  deriving stock (Show)++instance Exception LostConnection++-- | SYNAPSE-20: how a silent drop (no close, no reset: a peer that stopped,+-- a network that went away) is noticed. After 'kaIdle' seconds with+-- nothing heard, a ping is sent; when nothing (a pong or any message) is+-- heard 'kaTimeout' seconds after it, the connection is lost.+data Keepalive = Keepalive+  { kaIdle    :: Double+  , kaTimeout :: Double+  }+  deriving stock (Show, Eq)++-- | The keepalive: @PLEXUS_KEEPALIVE=IDLE[,TIMEOUT]@ in seconds, @0@ or+-- @off@ for none; by default a ping after 15 s of silence, lost 10 s later.+keepaliveFromEnv :: IO (Maybe Keepalive)+keepaliveFromEnv = do+  v <- lookupEnv "PLEXUS_KEEPALIVE"+  pure $ case fmap (T.splitOn "," . T.strip . T.pack) v of+    Nothing -> Just def+    Just [t] | t `elem` ["0", "off", "none"] -> Nothing+    Just [i] | Just i' <- num i, i' > 0 -> Just def { kaIdle = i' }+    Just [i, t] | Just i' <- num i, Just t' <- num t, i' > 0, t' > 0 -> Just (Keepalive i' t')+    Just _ -> Just def+  where+    def = Keepalive 15 10+    num :: Text -> Maybe Double+    num t = case readMaybe (T.unpack (T.strip t)) :: Maybe Double of+      Just d -> Just d+      Nothing -> fromIntegral <$> (readMaybe (T.unpack (T.strip t)) :: Maybe Int)++-- | SYNAPSE-20: why the connection is gone, if it is.+connectionLost :: SubstrateConnection -> IO (Maybe Text)+connectionLost c = atomically (tryReadTMVar (scClosed c))++-- | Say why the connection ended, from what ended the reader.+describeEnd :: SomeException -> Text+describeEnd e = case fromException e of+  Just (WS.CloseRequest code msg) ->+    "the server closed the connection (" <> T.pack (show code)+      <> (if LBS.null msg then "" else ": " <> T.pack (LBS.unpack msg)) <> ")"+  Just WS.ConnectionClosed -> "the connection was closed"+  Just other -> "the connection failed: " <> T.pack (show other)+  Nothing -> "the connection failed: " <> T.pack (show e)+ -- | Connect to substrate -- Throws an exception if connection fails connect :: SubstrateConfig -> IO SubstrateConnection@@ -92,51 +204,169 @@   nextId <- newIORef 1   subs   <- newTVarIO Map.empty   pendingReqs <- newTVarIO Map.empty+  closed <- newEmptyTMVarIO   resultVar <- newEmptyTMVarIO  -- Either error message or (conn, reader)+  ka <- keepaliveFromEnv+  -- SYNAPSE-20: when anything was last heard (a message or a pong)+  heard <- newIORef =<< getMonotonicTime+  let touch = getMonotonicTime >>= writeIORef heard+      lose why = atomically (void (tryPutTMVar closed why))+      opts = WS.defaultConnectionOptions { WS.connectionOnPong = touch }    -- Run WebSocket client in a background thread   -- Catch exceptions inside the thread to prevent them from being printed-  void $ forkIO $-    (WS.runClientWith substrateHost substratePort substratePath-        WS.defaultConnectionOptions substrateHeaders $ \conn -> do-      reader <- async $ readerLoop conn subs pendingReqs+  -- SCAFFOLD-21: over TLS when the config says wss+  let dial :: (Connection -> IO ()) -> IO ()+      dial = case substrateTls of+        Nothing -> WS.runClientWith substrateHost substratePort substratePath+                     opts substrateHeaders+        Just tls -> \app -> withTlsStream substrateHost substratePort tls $ \stream ->+          WS.runClientWithStream stream (hostHeader SubstrateConfig{..}) substratePath+            opts substrateHeaders app+  tid <- forkIO $+    (dial $ \conn -> do+      reader <- async $+        readerLoop conn touch subs pendingReqs+          `catch` (\(e :: SomeException) -> lose (describeEnd e))       -- Signal success       atomically $ putTMVar resultVar (Right (conn, reader))       -- Keep alive until reader exits       void (wait reader) `catch` \(_ :: SomeException) -> pure ()-    ) `catch` \(e :: SomeException) ->+    ) `catch` (\(e :: SomeException) ->       -- Signal failure (don't print, just capture)-      atomically $ putTMVar resultVar (Left $ show e)+      atomically $ void $ tryPutTMVar resultVar (Left $ show e))+      `finally` lose "the connection was closed"    -- Wait for connection result (with 5s timeout instead of hardcoded 200ms sleep)   mResult <- timeout 5000000 $ atomically $ takeTMVar resultVar    case mResult of-    Nothing -> error "Connection timeout (5s)"+    Nothing -> killThread tid >> error "Connection timeout (5s)"     Just (Left err) -> error $ "Connection failed: " <> err-    Just (Right (conn, reader)) ->+    Just (Right (conn, reader)) -> do+      kat <- traverse (\k -> async (keepaliveLoop k conn heard closed reader)) ka       pure SubstrateConnection         { scConnection    = conn         , scNextId        = nextId         , scSubscriptions = subs         , scPendingReqs   = pendingReqs         , scReaderThread  = reader+        , scClosed        = closed+        , scKeepalive     = kat         } +-- | SCAFFOLD-21: a TLS stream to @host:port@ for the action — the TCP+-- connection to the first address that answers, the handshake with the+-- name sent as SNI, the certificate chain verified against the system's+-- trust store and 'tlsCaFiles', and the name against the certificate. A+-- failed check fails the handshake (an exception, as a refused dial is).+withTlsStream :: String -> Int -> TlsOptions -> (WSS.Stream -> IO a) -> IO a+withTlsStream host port TlsOptions{..} act = do+  store <- trustStore tlsCaFiles+  let base = TLS.defaultParamsClient host (BS8.pack (show port))+      params = base+        { TLS.clientShared = (TLS.clientShared base) { TLS.sharedCAStore = store } }+  sock <- openTcp host port+  flip finally (S.close sock) $ do+    ctx <- TLS.contextNew sock params+    TLS.handshake ctx+    let recv = do+          chunk <- TLS.recvData ctx+          pure (if BS.null chunk then Nothing else Just chunk)+        send = maybe (TLS.bye ctx `catch` \(_ :: SomeException) -> pure ())+                     (TLS.sendData ctx)+    stream <- WSS.makeStream recv send+    act stream `finally` WSS.close stream++-- | The certificates a @wss://@ server's chain may end in: the system's+-- and those in each PEM file named. A file that holds none is an error+-- (a mistyped path would otherwise only show as a failed handshake).+trustStore :: [FilePath] -> IO CertificateStore+trustStore files = do+  system <- systemStore+  extra <- mapM readCa files+  pure (system <> makeCertificateStore (concat extra))+  where+    readCa f = do+      r <- try (readSignedObject f)+      case r of+        Left (e :: SomeException) -> ioError (userError ("CA file " <> f <> ": " <> show e))+        Right [] -> ioError (userError ("CA file " <> f <> " holds no PEM certificate"))+        Right certs -> pure certs++-- | The system's trust store, read once per process (every connection of+-- a pool would otherwise read the bundle again).+systemStore :: IO CertificateStore+systemStore = modifyMVar systemStoreMemo $ \case+  Just st -> pure (Just st, st)+  Nothing -> do+    st <- getSystemCertificateStore+    pure (Just st, st)++{-# NOINLINE systemStoreMemo #-}+systemStoreMemo :: MVar (Maybe CertificateStore)+systemStoreMemo = unsafePerformIO (newMVar Nothing)++-- | A TCP connection to the first of @host@'s addresses that takes it.+openTcp :: String -> Int -> IO S.Socket+openTcp host port = do+  let hints = S.defaultHints { S.addrSocketType = S.Stream }+  addrs <- S.getAddrInfo (Just hints) (Just host) (Just (show port))+  let attempt [] = ioError (userError ("no address of " <> host <> " answered"))+      attempt (a : rest) = do+        r <- try $ bracketOnError (S.openSocket a) S.close $ \s -> do+          S.setSocketOption s S.NoDelay 1+          S.connect s (S.addrAddress a)+          pure s+        case r of+          Right s -> pure s+          Left (e :: SomeException)+            | null rest -> throwIO e+            | otherwise -> attempt rest+  attempt addrs++-- | SYNAPSE-20: ping after 'kaIdle' seconds of silence; when nothing is+-- heard 'kaTimeout' seconds after the ping, the connection is lost — the+-- waiting calls are told and the reader is stopped, which closes the socket.+keepaliveLoop :: Keepalive -> Connection -> IORef Double -> TMVar Text -> Async () -> IO ()+keepaliveLoop Keepalive{..} conn heard closed reader = go Nothing+  where+    tick = min 1 (kaIdle / 2)+    go pinged = do+      threadDelay (round (tick * 1000000))+      gone <- atomically (not <$> isEmptyTMVar closed)+      unless gone $ do+        now <- getMonotonicTime+        h <- readIORef heard+        case pinged of+          Just at | h < at && now - at >= kaTimeout -> do+            atomically $ void $ tryPutTMVar closed $+              "no answer to a keepalive ping in " <> T.pack (show (round kaTimeout :: Int)) <> " s"+            cancel reader+          Just at | h < at -> go pinged+          _ | now - h >= kaIdle -> do+                WS.sendPing conn ("synapse" :: Text)+                  `catch` \(_ :: SomeException) -> pure ()+                go (Just now)+            | otherwise -> go Nothing+ -- | Disconnect from substrate disconnect :: SubstrateConnection -> IO () disconnect SubstrateConnection{..} = do+  mapM_ cancel scKeepalive   cancel scReaderThread   WS.sendClose scConnection ("bye" :: Text) `catch` \(_ :: SomeException) -> pure ()  -- | Reader loop - dispatches incoming messages to the right handler readerLoop   :: Connection+  -> IO ()   -> TVar (Map SubscriptionId (TQueue Value))   -> TVar (Map RequestId PendingRequest)   -> IO ()-readerLoop conn subs pendingReqs = forever $ do+readerLoop conn touch subs pendingReqs = forever $ do   msg <- WS.receiveData conn+  touch   case eitherDecode msg of     Left err -> putStrLn $ "Failed to decode message: " <> err     Right val -> dispatch val@@ -212,7 +442,24 @@   -> Text              -- ^ Method name (e.g., "bash_execute")   -> Value             -- ^ Parameters   -> Stream (Of PlexusStreamItem) IO ()-substrateRpc conn method params = do+substrateRpc conn = substrateRpcWith conn (pure ())++-- | SYNAPSE-20: 'substrateRpc', telling @onSubscribed@ when the server has+-- taken the call (its subscription confirmed) — before any item.+substrateRpcWith+  :: SubstrateConnection+  -> IO ()+  -> Text+  -> Value+  -> Stream (Of PlexusStreamItem) IO ()+substrateRpcWith conn onSubscribed method params = do+  -- SYNAPSE-20: a connection already gone (it died idle in a pool) takes+  -- no request; nothing was sent, so the caller may always try again+  gone <- liftIO $ connectionLost conn+  case gone of+    Just why -> liftIO $ throwIO (LostConnection why False)+    Nothing -> pure ()+   -- Get next request ID   rid <- liftIO $ atomicModifyIORef' (scNextId conn) $ \n -> (n + 1, RequestId n) @@ -224,27 +471,34 @@   liftIO $ atomically $ modifyTVar' (scPendingReqs conn) $     Map.insert rid (PendingRequest queue respVar) -  -- Send subscription request+  -- Send subscription request (a send that fails sent nothing)   let req = mkSubscribeRequest rid method params   liftIO $ WS.sendTextData (scConnection conn) (encode req)+    `catch` \(e :: SomeException) -> do+      why <- maybe (describeEnd e) id <$> connectionLost conn+      throwIO (LostConnection why False) -  -- Wait for subscription confirmation-  resp <- liftIO $ atomically $ takeTMVar respVar+  -- Wait for subscription confirmation — or for the connection to go+  resp <- liftIO $ atomically $+    (Right <$> takeTMVar respVar) `orElse` (Left <$> readTMVar (scClosed conn))    -- Clean up pending request   liftIO $ atomically $ modifyTVar' (scPendingReqs conn) $ Map.delete rid    case resp of-    RpcError _ err -> do+    Left why -> liftIO $ throwIO (LostConnection why True)++    Right (RpcError _ err) -> do       -- Emit error as StreamError so caller can handle it properly-      Str.yield $ StreamError+      Str.yield $ StreamErrorWith         { itemPlexusHash = ""  -- No hash available for subscription errors         , itemProvenance = Provenance ["substrate"]         , itemError = T.pack $ "Subscription error: " <> show err         , itemRecoverable = False+        , itemErrorRefusal = Nothing         } -    RpcSuccess _ result -> do+    Right (RpcSuccess _ result) -> do       -- The reader loop already registered the queue in scSubscriptions       -- when it processed the response, so we just need the subId for cleanup       case fromJSON result of@@ -253,12 +507,22 @@           pure ()          Success subId -> do+          liftIO onSubscribed           -- Stream items until done, cleanup when finished           streamItems queue subId <* liftIO (cleanup subId)   where+    -- SYNAPSE-20: what arrived before the connection went is delivered+    -- first ('orElse' tries the queue first); then the call fails with+    -- 'LostConnection' instead of waiting forever+    next :: TQueue Value -> IO Value+    next queue = do+      r <- atomically $+        (Right <$> readTQueue queue) `orElse` (Left <$> readTMVar (scClosed conn))+      either (\why -> throwIO (LostConnection why True)) pure r+     streamItems :: TQueue Value -> SubscriptionId -> Stream (Of PlexusStreamItem) IO ()     streamItems queue subId = do-      val <- liftIO $ atomically $ readTQueue queue+      val <- liftIO $ next queue       case fromJSON val of         Error err -> do           liftIO $ putStrLn $ "Failed to parse stream item: " <> err

@@ -39,6 +39,15 @@   , ParamSchema(..)   , MethodRole(..) +    -- * The caller's view (AUTH-3 / AUTH-4)+  , View(..)+  , hasAuthColumns++    -- * Help metadata (SYNAPSE-13)+  , MethodExample(..)+  , paramValues+  , xPlexusValues+     -- * Queries   , isHubActivation   , isLeafActivation@@ -55,11 +64,16 @@  import Data.Aeson import Data.Aeson.Types (Parser)+import qualified Data.Aeson.Key as K+import qualified Data.Aeson.KeyMap as KM import Data.Maybe (catMaybes, fromMaybe) import Data.Text (Text) import qualified Data.Text as T+import Data.Map.Strict (Map) import GHC.Generics (Generic) +import Plexus.Auth (View(..))+ -- ============================================================================ -- Deprecation Metadata (IR-5) -- ============================================================================@@ -182,8 +196,9 @@ instance FromJSON ChildSummary where   parseJSON = withObject "ChildSummary" $ \o -> ChildSummary     <$> o .: "namespace"-    <*> o .: "description"-    <*> o .: "hash"+    -- AUTH-4: an anonymous view's children may be bare+    <*> o .:? "description" .!= ""+    <*> o .:? "hash" .!= ""  instance ToJSON ChildSummary where   toJSON ChildSummary{..} = object@@ -206,14 +221,22 @@   , methodDeprecation     :: Maybe DeprecationInfo  -- ^ Deprecation info, if any (IR-5)   , methodParamSchemas    :: Maybe [ParamSchema]    -- ^ Optional structured param info with per-field deprecations (IR-5)   , methodRole            :: MethodRole             -- ^ Structural role in the activation graph (IR-2 / IR-3); defaults to 'MethodRoleRpc'+  , methodLongDescription :: Maybe Text             -- ^ AUTH-3: the doc comment, beside the short description+  , methodGrant           :: Maybe Text             -- ^ AUTH-3: @<activation>.<method>@, e.g. @tickets.set_status@+  , methodTarget          :: Maybe Text             -- ^ AUTH-3: the param that locates what a call acts on+  , methodHeldOn          :: Maybe [Text]           -- ^ AUTH-4: where the caller holds the grant (its view only)+  , methodPublic          :: Bool                   -- ^ AUTH-3: callable anonymously (the auth activation's door)+  , methodGroup           :: Maybe Text             -- ^ SYNAPSE-13: the method's group in a listing (OpenAPI's tag)+  , methodExamples        :: [MethodExample]        -- ^ SYNAPSE-13: worked calls, each its params and a note   }   deriving stock (Show, Eq, Generic)  instance FromJSON MethodSchema where   parseJSON = withObject "MethodSchema" $ \o -> MethodSchema     <$> o .: "name"-    <*> o .: "description"-    <*> o .: "hash"+    -- AUTH-4: an anonymous view may name a method and no more+    <*> o .:? "description" .!= ""+    <*> o .:? "hash" .!= ""     <*> o .:? "params"     <*> o .:? "returns"     <*> o .:? "streaming"      .!= False@@ -223,9 +246,17 @@     <*> o .:? "deprecation"     <*> o .:? "param_schemas"     <*> o .:? "role" .!= MethodRoleRpc+    <*> o .:? "long_description"+    <*> o .:? "grant"+    <*> o .:? "target"+    <*> o .:? "held_on"+    <*> o .:? "public" .!= False+    -- SYNAPSE-13: absent on an older hub; an unreadable one is ignored+    <*> (o .:? "group" <|> pure Nothing)+    <*> (o .:? "examples" .!= [] <|> pure [])  instance ToJSON MethodSchema where-  toJSON MethodSchema{..} = object+  toJSON MethodSchema{..} = object $     [ "name"           .= methodName     , "description"    .= methodDescription     , "hash"           .= methodHash@@ -238,8 +269,46 @@     , "deprecation"    .= methodDeprecation     , "param_schemas"  .= methodParamSchemas     , "role"           .= methodRole+    ] <> catMaybes+    [ ("long_description" .=) <$> methodLongDescription+    , ("grant"            .=) <$> methodGrant+    , ("target"           .=) <$> methodTarget+    , ("held_on"          .=) <$> methodHeldOn+    , if methodPublic then Just ("public" .= True) else Nothing+    , ("group"            .=) <$> methodGroup+    , if null methodExamples then Nothing else Just ("examples" .= methodExamples)     ] +-- | SYNAPSE-13: one worked call of a method — the params it sends and a+-- note saying what it does. A client renders it as a command line.+data MethodExample = MethodExample+  { exampleParams :: Value       -- ^ the params object, as the method takes it+  , exampleNote   :: Maybe Text  -- ^ what the call does, in a line+  }+  deriving stock (Show, Eq, Generic)++instance FromJSON MethodExample where+  parseJSON = withObject "MethodExample" $ \o -> MethodExample+    <$> o .:? "params" .!= Object mempty+    <*> o .:? "note"++instance ToJSON MethodExample where+  toJSON MethodExample{..} = object $+    ("params" .= exampleParams) : maybe [] (\n -> ["note" .= n]) exampleNote++-- | SYNAPSE-13: the JSON Schema extension key on a param's property that+-- lists the values it is meant to take — a hint, never enforced.+xPlexusValues :: Text+xPlexusValues = "x-plexus-values"++-- | SYNAPSE-13: the values a param's property says it is meant to take+-- (@x-plexus-values@), when it says them.+paramValues :: Value -> Maybe [Text]+paramValues (Object po) = case KM.lookup (K.fromText xPlexusValues) po of+  Just (Array vs) -> Just [t | String t <- foldr (:) [] vs]+  _ -> Nothing+paramValues _ = Nothing+ -- | Shallow plugin schema (what we receive from {backend}.schema) -- -- Children are summaries only - fetch full schema on-demand when navigating.@@ -253,22 +322,27 @@   , psMethods         :: [MethodSchema]   , psChildren        :: Maybe [ChildSummary]  -- ^ Nothing = leaf, Just = hub activation   , psDeprecation     :: Maybe DeprecationInfo -- ^ Activation-level deprecation (IR-5)+  , psRoles           :: Maybe (Map Text [Text]) -- ^ AUTH-3: the activation's default roles, each its grants+  , psView            :: Maybe View            -- ^ AUTH-4: which view this schema answers from   }   deriving stock (Show, Eq, Generic)  instance FromJSON PluginSchema where   parseJSON = withObject "PluginSchema" $ \o -> PluginSchema     <$> o .: "namespace"-    <*> o .: "version"-    <*> o .: "description"+    -- AUTH-4: an anonymous view's schema may omit what it does not show+    <*> o .:? "version" .!= ""+    <*> o .:? "description" .!= ""     <*> o .:? "long_description"-    <*> o .: "hash"+    <*> o .:? "hash" .!= ""     <*> o .:? "methods" .!= []     <*> o .:? "children"     <*> o .:? "deprecation"+    <*> (o .:? "roles" <|> pure Nothing)+    <*> (o .:? "view" <|> pure Nothing)  instance ToJSON PluginSchema where-  toJSON PluginSchema{..} = object+  toJSON PluginSchema{..} = object $     [ "namespace"        .= psNamespace     , "version"          .= psVersion     , "description"      .= psDescription@@ -277,6 +351,9 @@     , "methods"          .= psMethods     , "children"         .= psChildren     , "deprecation"      .= psDeprecation+    ] <> catMaybes+    [ ("roles" .=) <$> psRoles+    , ("view"  .=) <$> psView     ]  -- | Result of a schema query - can be either a full plugin or just a method@@ -299,6 +376,11 @@ -- ============================================================================ -- Basic Queries -- ============================================================================++-- | AUTH-4: does any method here say its grant or where it is held? Then+-- help shows those columns; otherwise it renders as before AUTH.+hasAuthColumns :: PluginSchema -> Bool+hasAuthColumns ps = any (\m -> maybe False (const True) (methodGrant m) || maybe False (const True) (methodHeldOn m)) (psMethods ps)  -- | Is this a hub activation (has children)? isHubActivation :: PluginSchema -> Bool

@@ -10,6 +10,7 @@     -- * RPC Calls (streaming)   , rpcCallStreaming   , invokeMethodStreaming+  , invokeMethodStreamingWith      -- * Bidirectional Response   , sendBidirectionalResponse@@ -37,7 +38,7 @@ import qualified Network.Socket as NS import System.IO.Unsafe (unsafePerformIO) -import Plexus.Client (SubstrateConfig(..), connect, disconnect, substrateRpc, defaultConfig)+import Plexus.Client (SubstrateConfig(..), LostConnection(..), connect, disconnect, substrateRpc, substrateRpcWith, defaultConfig) import Plexus.Client.Pool (ConnectionPool, createConnectionPool, withPooledConnection, defaultPlexusPoolConfig) import Plexus.Types (PlexusStreamItem(..), TransportError(..), Response(..), StandardResponse) import Plexus.Schema.Recursive (PluginSchema, MethodSchema, SchemaResult(..), parsePluginSchema, parseSchemaResult)@@ -71,6 +72,9 @@ -- | Categorize exceptions into typed TransportError categorizeException :: SubstrateConfig -> SomeException -> IO (Either TransportError a) categorizeException cfg e+  -- SYNAPSE-20: the connection went while the call was on it+  | Just (LostConnection why sent) <- fromException e =+      pure $ Left $ ConnectionLost why sent   -- Check for connection refused (most common)   | Just ioErr <- fromException e :: Maybe IOException = do       let errMsg = show ioErr@@ -87,7 +91,14 @@       let errMsg = show e           host = T.pack $ substrateHost cfg           port = substratePort cfg-      in pure $ Left $ if "protocol" `isInfixOf` errMsg || "parse" `isInfixOf` errMsg+      in pure $ Left $+        -- SYNAPSE-20: 'connect' reports a refused or timed-out dial as an+        -- error call; say which, so a caller can tell an outage from a fault+        if "refused" `isInfixOf` errMsg || "ECONNREFUSED" `isInfixOf` errMsg+          then ConnectionRefused host port+        else if "Connection timeout" `isInfixOf` errMsg+          then ConnectionTimeout host port+        else if "protocol" `isInfixOf` errMsg || "parse" `isInfixOf` errMsg         then ProtocolError (T.pack errMsg)         else NetworkError (T.pack errMsg) @@ -96,11 +107,19 @@     on f g x y = f (g x) (g y)  doCallInner :: SubstrateConfig -> Text -> Value -> IO [PlexusStreamItem]-doCallInner cfg method params = do+doCallInner cfg method params = freshIfUnsent $ do   pool <- getOrCreatePool cfg   withPooledConnection pool $ \conn ->     S.toList_ $ substrateRpc conn method params +-- | SYNAPSE-20: a pooled connection that died while idle fails the call+-- before anything is sent (and the pool drops it); the call goes once more,+-- on a fresh connection. A call whose request went out is never repeated+-- here — whether that is safe is the caller's to say.+freshIfUnsent :: IO a -> IO a+freshIfUnsent act = act `E.catch` \(e :: LostConnection) ->+  if lostSent e then E.throwIO e else act+ -- | Streaming RPC call - invokes callback for each item as it arrives rpcCallStreaming :: SubstrateConfig -> Text -> Value -> (PlexusStreamItem -> IO ()) -> IO (Either TransportError ()) rpcCallStreaming cfg method params onItem = do@@ -109,19 +128,28 @@   pure result  doCallStreaming :: SubstrateConfig -> Text -> Value -> (PlexusStreamItem -> IO ()) -> IO ()-doCallStreaming cfg method params onItem = do+doCallStreaming cfg = doCallStreamingWith cfg (pure ())++doCallStreamingWith :: SubstrateConfig -> IO () -> Text -> Value -> (PlexusStreamItem -> IO ()) -> IO ()+doCallStreamingWith cfg onSubscribed method params onItem = freshIfUnsent $ do   pool <- getOrCreatePool cfg   withPooledConnection pool $ \conn ->-    S.mapM_ onItem $ substrateRpc conn method params+    S.mapM_ onItem $ substrateRpcWith conn onSubscribed method params  -- | Streaming method invocation invokeMethodStreaming :: SubstrateConfig -> [Text] -> Text -> Value -> (PlexusStreamItem -> IO ()) -> IO (Either TransportError ())-invokeMethodStreaming cfg namespacePath method params onItem = do+invokeMethodStreaming cfg = invokeMethodStreamingWith cfg (pure ())++-- | SYNAPSE-20: 'invokeMethodStreaming', telling @onSubscribed@ when the+-- server has taken the call, before any item.+invokeMethodStreamingWith :: SubstrateConfig -> IO () -> [Text] -> Text -> Value -> (PlexusStreamItem -> IO ()) -> IO (Either TransportError ())+invokeMethodStreamingWith cfg onSubscribed namespacePath method params onItem = do   let backend = substrateBackend cfg   let fullPath = if null namespacePath then [backend] else namespacePath   let dotPath = T.intercalate "." (fullPath ++ [method])   let callParams = object ["method" .= dotPath, "params" .= params]-  rpcCallStreaming cfg (backend <> ".call") callParams onItem+  (Right <$> doCallStreamingWith cfg onSubscribed (backend <> ".call") callParams onItem)+    `catch` categorizeException cfg  -- | Fetch schema at a specific path -- Empty path = root (<backend>.schema)

@@ -1,3 +1,5 @@+{-# LANGUAGE PatternSynonyms #-}+ -- | JSON-RPC types for communicating with Plexus RPC servers module Plexus.Types   ( -- * JSON-RPC Protocol@@ -10,7 +12,9 @@   , SubscriptionId(..)      -- * Plexus Stream Types-  , PlexusStreamItem(..)+  , PlexusStreamItem(.., StreamError)+  , itemRefusal+  , Refusal(..)   , Provenance(..)   , StreamMetadata(..)   , GuidanceErrorType(..)@@ -32,7 +36,9 @@   , getPlexusHash   ) where +import Control.Applicative ((<|>)) import Data.Aeson+import qualified Data.Aeson.KeyMap as KM import Data.Aeson.Types (Parser) import qualified Data.ByteString.Lazy as LBS import Data.Maybe (catMaybes)@@ -41,6 +47,8 @@ import qualified Data.Text.Encoding as T import GHC.Generics (Generic) +import Plexus.Auth (Refusal(..), refusalFromObject)+ -- | Request ID for JSON-RPC calls newtype RequestId = RequestId { unRequestId :: Int }   deriving stock (Show, Eq, Ord)@@ -283,6 +291,15 @@   | NetworkError       { transportMessage :: Text       }+  -- | SYNAPSE-20: an open connection was lost (the server closed it, it was+  -- reset, or a keepalive ping went unanswered) while a call was on it.+  -- 'transportSent' says whether the call's request had been sent: when it+  -- had not, repeating the call is always safe; when it had, the server+  -- may have acted on it.+  | ConnectionLost+      { transportMessage :: Text+      , transportSent    :: Bool+      }   deriving stock (Show, Eq, Generic)  instance ToJSON TransportError where@@ -304,6 +321,11 @@     [ "error_kind" .= ("network_error" :: Text)     , "message" .= msg     ]+  toJSON (ConnectionLost msg sent) = object+    [ "error_kind" .= ("connection_lost" :: Text)+    , "message" .= msg+    , "sent" .= sent+    ]  instance FromJSON TransportError where   parseJSON = withObject "TransportError" $ \o -> do@@ -313,6 +335,7 @@       "connection_timeout" -> ConnectionTimeout <$> o .: "host" <*> o .: "port"       "protocol_error" -> ProtocolError <$> o .: "message"       "network_error" -> NetworkError <$> o .: "message"+      "connection_lost" -> ConnectionLost <$> o .: "message" <*> o .: "sent"       _ -> fail $ "Unknown transport error_kind: " <> T.unpack kind  -- ============================================================================@@ -463,11 +486,14 @@       , itemAvailableMethods  :: Maybe [Text]       , itemMethodSchema      :: Maybe Value       }-  | StreamError+  | StreamErrorWith       { itemPlexusHash  :: Text       , itemProvenance  :: Provenance       , itemError       :: Text       , itemRecoverable :: Bool+      , itemErrorRefusal :: Maybe Refusal+        -- ^ AUTH-2: @error_kind: "forbidden"@ and what it names; Nothing+        -- for every other error (and on every hub without a gate)       }   | StreamRequest       { itemPlexusHash  :: Text@@ -482,6 +508,24 @@       }   deriving stock (Show, Eq, Generic) +-- | An error item as it was before AUTH-2: hash, provenance, message,+-- recoverable. Matches every error (a refusal too); builds one with no+-- refusal. 'StreamErrorWith' carries the typed refusal.+pattern StreamError :: Text -> Provenance -> Text -> Bool -> PlexusStreamItem+pattern StreamError hash prov err rec <- StreamErrorWith hash prov err rec _+  where StreamError hash prov err rec = StreamErrorWith hash prov err rec Nothing++{-# COMPLETE StreamProgress, StreamData, StreamGuidance, StreamError, StreamRequest, StreamDone #-}++-- | AUTH-2: the refusal an item carries — an error item with+-- @error_kind: "forbidden"@, or a data item whose content is one (an+-- in-band error event naming the same fields).+itemRefusal :: PlexusStreamItem -> Maybe Refusal+itemRefusal = \case+  StreamErrorWith _ _ _ _ r -> r+  StreamData _ _ _ (Object o) -> refusalFromObject o+  _ -> Nothing+ instance FromJSON PlexusStreamItem where   parseJSON = withObject "PlexusStreamItem" $ \o -> do     typ <- o .: "type" :: Parser Text@@ -509,9 +553,10 @@             <*> o .: "suggestion"             <*> o .:? "available_methods"             <*> o .:? "method_schema"-          "error" -> StreamError hash prov+          "error" -> StreamErrorWith hash prov             <$> o .: "message"             <*> o .:? "recoverable" .!= False+            <*> pure (refusalFromObject o)           "request" -> StreamRequest hash prov             <$> o .: "request_id"             <*> o .: "request"@@ -522,7 +567,9 @@       -- Legacy format: flat structure       parseLegacy :: Text -> Object -> Parser PlexusStreamItem       parseLegacy typ o = do-        hash <- o .: "plexus_hash"+        -- an error item may come bare (AUTH-2's refusal is written so):+        -- no hash or provenance is no reason to lose it+        hash <- if typ == "error" then o .:? "plexus_hash" .!= "" else o .: "plexus_hash"         case typ of           "progress" -> StreamProgress hash             <$> o .: "provenance"@@ -538,10 +585,11 @@             <*> o .: "suggestion"             <*> o .:? "available_methods"             <*> o .:? "method_schema"-          "error" -> StreamError hash-            <$> o .: "provenance"-            <*> o .: "error"-            <*> o .: "recoverable"+          "error" -> StreamErrorWith hash+            <$> o .:? "provenance" .!= Provenance []+            <*> (o .: "error" <|> o .: "message")+            <*> o .:? "recoverable" .!= False+            <*> pure (refusalFromObject o)           "request" -> StreamRequest hash             <$> o .: "provenance"             <*> o .: "request_id"@@ -573,12 +621,17 @@     [ ("available_methods" .=) <$> availMethods     , ("method_schema" .=) <$> methodSchema     ]-  toJSON (StreamError hash prov err rec) = object+  toJSON (StreamErrorWith hash prov err rec refusal) = object $     [ "type" .= ("error" :: Text)     , "metadata" .= StreamMetadata prov hash 0     , "message" .= err     , "recoverable" .= rec-    ]+    ] <> refusalFields+    where+      -- the refusal's own fields, beside the message (as the hub sent them)+      refusalFields = case toJSON <$> refusal of+        Just (Object ro) -> [ (k, v) | (k, v) <- KM.toList ro, k /= "message" ]+        _ -> []   toJSON (StreamRequest hash prov reqId reqData timeout) = object $     [ "type" .= ("request" :: Text)     , "metadata" .= StreamMetadata prov hash 0