diff --git a/plexus-protocol.cabal b/plexus-protocol.cabal --- a/plexus-protocol.cabal +++ b/plexus-protocol.cabal @@ -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, diff --git a/src/Plexus.hs b/src/Plexus.hs --- a/src/Plexus.hs +++ b/src/Plexus.hs @@ -34,6 +34,8 @@ ( -- * Connection (re-exported from Plexus.Client) SubstrateConnection , SubstrateConfig(..) + , TlsOptions(..) + , defaultTlsOptions , defaultConfig , connect , disconnect diff --git a/src/Plexus/Auth.hs b/src/Plexus/Auth.hs new file mode 100644 --- /dev/null +++ b/src/Plexus/Auth.hs @@ -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 diff --git a/src/Plexus/Client.hs b/src/Plexus/Client.hs --- a/src/Plexus/Client.hs +++ b/src/Plexus/Client.hs @@ -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 diff --git a/src/Plexus/Schema/Recursive.hs b/src/Plexus/Schema/Recursive.hs --- a/src/Plexus/Schema/Recursive.hs +++ b/src/Plexus/Schema/Recursive.hs @@ -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: @.@, 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 diff --git a/src/Plexus/Transport.hs b/src/Plexus/Transport.hs --- a/src/Plexus/Transport.hs +++ b/src/Plexus/Transport.hs @@ -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 (.schema) diff --git a/src/Plexus/Types.hs b/src/Plexus/Types.hs --- a/src/Plexus/Types.hs +++ b/src/Plexus/Types.hs @@ -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