feat(dgw): subscriber API

README.md

@@ -49,6 +49,11 @@ Currently stable options are:
 
     Host segment may be abridged with `*`.
 
+- `Subscriber`: subscriber configuration:
+
+    * `Url`: HTTP URL where notification messages are to be sent,
+    * `Token`: bearer token to use when making HTTP requests.
+
 ## Sample Usage
 
 ### RDP routing

devolutions-gateway/doc/api.yaml

@@ -4,7 +4,7 @@ info:
   description: Protocol-aware fine-grained relay server
   contact:
     name: Devolutions Inc.
-    email: ''
+    email: [email protected]
   license:
     name: MIT/Apache-2.0
   version: 2022.2.2
@@ -241,6 +241,8 @@ components:
           format: uuid
         SubProvisionerPublicKey:
           $ref: '#/components/schemas/SubProvisionerKey'
+        Subscriber:
+          $ref: '#/components/schemas/Subscriber'
     ConnectionMode:
       type: string
       enum:
@@ -267,7 +269,7 @@ components:
         jti:
           type: string
           format: uuid
-          description: Unique ID for current  JRL
+          description: Unique ID for current JRL
     ListenerUrls:
       type: object
       required:
@@ -323,4 +325,25 @@ components:
           type: string
         Value:
           type: string
+    Subscriber:
+      type: object
+      required:
+      - Url
+      - Token
+      properties:
+        Token:
+          type: string
+        Url:
+          type: string
+  securitySchemes:
+    jrl_token:
+      type: http
+      scheme: bearer
+      bearerFormat: JWT
+      description: Contains the JRL to apply if newer
+    scope_token:
+      type: http
+      scheme: bearer
+      bearerFormat: JWT
+      description: Token allowing a single HTTP request for a specific scope
 

devolutions-gateway/doc/subscriber-api.yaml

@@ -0,0 +1,78 @@
+openapi: 3.0.3
+info:
+  title: devolutions-gateway-subscriber
+  description: API a service must implement in order to receive Devolutions Gateway
+    notifications
+  contact:
+    name: Devolutions Inc.
+    email: [email protected]
+  license:
+    name: MIT/Apache-2.0
+  version: 2022.2.2
+paths:
+  /:
+    post:
+      tags:
+      - crate::subscriber
+      summary: Process a message originating from a Devolutions Gateway instance
+      description: |
+        Process a message originating from a Devolutions Gateway instance
+      operationId: PostMessage
+      requestBody:
+        description: Message
+        content:
+          application/json:
+            schema:
+              $ref: '#/components/schemas/SubscriberMessage'
+        required: true
+      responses:
+        '200':
+          description: Message received and processed successfuly
+        '400':
+          description: Bad message
+        '401':
+          description: Invalid or missing authorization token
+        '403':
+          description: Insufficient permissions
+      deprecated: false
+      security:
+      - subscriber_token: []
+components:
+  schemas:
+    SubscriberMessage:
+      type: object
+      required:
+      - kind
+      properties:
+        kind:
+          $ref: '#/components/schemas/SubscriberMessageKind'
+        session:
+          $ref: '#/components/schemas/SubscriberSessionInfo'
+        session_list:
+          type: array
+          items:
+            $ref: '#/components/schemas/SubscriberSessionInfo'
+    SubscriberMessageKind:
+      type: string
+      enum:
+      - session.started
+      - session.ended
+      - session.list
+    SubscriberSessionInfo:
+      type: object
+      required:
+      - association_id
+      - start_timestamp
+      properties:
+        association_id:
+          type: string
+          format: uuid
+        start_timestamp:
+          type: string
+          format: date-time
+  securitySchemes:
+    subscriber_token:
+      type: http
+      scheme: bearer
+      description: Token allowing to push messages
+

devolutions-gateway/src/config.rs

@@ -13,6 +13,7 @@ use std::fs::File;
 use std::io::BufReader;
 use std::sync::Arc;
 use tap::prelude::*;
+use tokio::sync::Notify;
 use tokio_rustls::rustls;
 use url::Url;
 use uuid::Uuid;
@@ -80,8 +81,9 @@ impl Tls {
 #[derive(Debug, Clone)]
 pub struct Conf {
     pub id: Option<Uuid>,
-    pub listeners: Vec<ListenerUrls>,
     pub hostname: String,
+    pub listeners: Vec<ListenerUrls>,
+    pub subscriber: Option<dto::Subscriber>,
     pub capture_path: Option<Utf8PathBuf>,
     pub log_file: Utf8PathBuf,
     pub log_directive: Option<String>,
@@ -176,8 +178,9 @@ impl Conf {
 
         Ok(Conf {
             id: conf_file.id,
-            listeners,
             hostname,
+            listeners,
+            subscriber: conf_file.subscriber.clone(),
             capture_path: conf_file.capture_path.clone(),
             log_file,
             log_directive: conf_file.log_directive.clone(),
@@ -203,6 +206,7 @@ pub struct ConfHandle {
 struct ConfHandleInner {
     conf: parking_lot::RwLock<Arc<Conf>>,
     conf_file: parking_lot::RwLock<Arc<dto::ConfFile>>,
+    changed: Notify,
 }
 
 impl ConfHandle {
@@ -228,6 +232,7 @@ impl ConfHandle {
             inner: Arc::new(ConfHandleInner {
                 conf: parking_lot::RwLock::new(Arc::new(conf)),
                 conf_file: parking_lot::RwLock::new(Arc::new(conf_file)),
+                changed: Notify::new(),
             }),
         })
     }
@@ -242,13 +247,19 @@ impl ConfHandle {
         self.inner.conf_file.read().clone()
     }
 
+    /// Waits for configuration to be changed
+    pub async fn change_notified(&self) {
+        self.inner.changed.notified().await;
+    }
+
     /// Atomically saves and replaces current configuration with a new one
     #[instrument(skip(self))]
     pub fn save_new_conf_file(&self, conf_file: dto::ConfFile) -> anyhow::Result<()> {
         let conf = Conf::from_conf_file(&conf_file).context("Invalid configuration file")?;
         save_config(&conf_file).context("Failed to save configuration")?;
         *self.inner.conf.write() = Arc::new(conf);
         *self.inner.conf_file.write() = Arc::new(conf_file);
+        self.inner.changed.notify_waiters();
         trace!("success");
         Ok(())
     }
@@ -503,6 +514,10 @@ pub mod dto {
         /// Listeners to launch at startup
         pub listeners: Vec<ListenerConf>,
 
+        /// Subscriber API
+        #[serde(skip_serializing_if = "Option::is_none")]
+        pub subscriber: Option<Subscriber>,
+
         /// (Unstable) Folder and prefix for log files
         #[serde(skip_serializing_if = "Option::is_none")]
         pub log_file: Option<Utf8PathBuf>,
@@ -561,6 +576,7 @@ pub mod dto {
                         external_url: "wss://*:7171".try_into().unwrap(),
                     },
                 ],
+                subscriber: None,
                 log_file: None,
                 jrl_file: None,
                 log_directive: None,
@@ -735,4 +751,13 @@ pub mod dto {
         pub internal_url: String,
         pub external_url: String,
     }
+
+    #[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
+    #[derive(PartialEq, Eq, Debug, Clone, Serialize, Deserialize)]
+    #[serde(rename_all = "PascalCase")]
+    pub struct Subscriber {
+        #[cfg_attr(feature = "openapi", schema(value_type = String))]
+        pub url: Url,
+        pub token: String,
+    }
 }

devolutions-gateway/src/generic_client.rs

@@ -2,6 +2,7 @@ use crate::config::Conf;
 use crate::jet_client::JetAssociationsMap;
 use crate::preconnection_pdu::{extract_association_claims, read_preconnection_pdu};
 use crate::rdp::RdpClient;
+use crate::subscriber::SubscriberSender;
 use crate::token::{ApplicationProtocol, ConnectionMode, CurrentJrl, Protocol, TokenCache};
 use crate::{utils, ConnectionModeDetails, GatewaySessionInfo, Proxy};
 use anyhow::Context;
@@ -14,37 +15,40 @@ use typed_builder::TypedBuilder;
 
 #[derive(TypedBuilder)]
 pub struct GenericClient {
-    config: Arc<Conf>,
+    conf: Arc<Conf>,
     associations: Arc<JetAssociationsMap>,
     token_cache: Arc<TokenCache>,
     jrl: Arc<CurrentJrl>,
     client_addr: SocketAddr,
     client_stream: TcpStream,
+    subscriber_tx: SubscriberSender,
 }
 
 impl GenericClient {
     pub async fn serve(self) -> anyhow::Result<()> {
         let Self {
-            config,
+            conf,
             associations,
             token_cache,
             jrl,
             client_addr,
             mut client_stream,
+            subscriber_tx,
         } = self;
 
         let (pdu, mut leftover_bytes) = read_preconnection_pdu(&mut client_stream).await?;
         let source_ip = client_addr.ip();
-        let association_claims = extract_association_claims(&pdu, source_ip, &config, &token_cache, &jrl)?;
+        let association_claims = extract_association_claims(&pdu, source_ip, &conf, &token_cache, &jrl)?;
 
         match association_claims.jet_ap {
             // We currently special case this because it may be the "RDP-TLS" protocol
             ApplicationProtocol::Known(Protocol::Rdp) => {
                 RdpClient {
-                    config,
+                    conf,
                     associations,
                     token_cache,
                     jrl,
+                    subscriber_tx,
                 }
                 .serve_with_association_claims_and_leftover_bytes(
                     client_addr,
@@ -72,6 +76,7 @@ impl GenericClient {
                             .associations(associations)
                             .association_id(association_id)
                             .client_transport(AnyStream::from(client_stream))
+                            .subscriber_tx(subscriber_tx)
                             .build()
                             .start(&leftover_bytes)
                             .await
@@ -105,10 +110,11 @@ impl GenericClient {
                         .with_filtering_policy(filtering_policy);
 
                         Proxy::init()
-                            .config(config)
+                            .conf(conf)
                             .session_info(info)
                             .addrs(client_addr, server_transport.addr)
                             .transports(client_stream, server_transport)
+                            .subscriber(subscriber_tx)
                             .select_dissector_and_forward()
                             .await
                             .context("Encountered a failure during plain tcp traffic proxying")

devolutions-gateway/src/http/controllers/config.rs

@@ -1,4 +1,4 @@
-use crate::config::dto::{DataEncoding, PubKeyFormat};
+use crate::config::dto::{DataEncoding, PubKeyFormat, Subscriber};
 use crate::config::ConfHandle;
 use crate::http::guards::access::{AccessGuard, TokenType};
 use crate::http::HttpErrorStatus;
@@ -29,8 +29,6 @@ impl ConfigController {
     }
 }
 
-const KEY_ALLOWLIST: &[&str] = &["Id", "SubProvisionerPublicKey"];
-
 #[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
 #[derive(Serialize, Deserialize)]
 #[serde(rename_all = "PascalCase")]
@@ -39,6 +37,8 @@ pub struct ConfigPatch {
     pub id: Option<Uuid>,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub sub_provisioner_public_key: Option<SubProvisionerKey>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub subscriber: Option<Subscriber>,
 }
 
 #[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
@@ -51,6 +51,8 @@ pub struct SubProvisionerKey {
     pub encoding: Option<DataEncoding>,
 }
 
+const KEY_ALLOWLIST: &[&str] = &["Id", "SubProvisionerPublicKey", "Subscriber"];
+
 /// Modifies configuration
 #[cfg_attr(feature = "openapi", utoipa::path(
     patch,

devolutions-gateway/src/http/controllers/jrl.rs

@@ -121,7 +121,7 @@ async fn get_jrl_info(revocation_list: &CurrentJrl) -> Json<JrlInfo> {
 #[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
 #[derive(Serialize)]
 pub struct JrlInfo {
-    /// Unique ID for current  JRL
+    /// Unique ID for current JRL
     pub jti: Uuid,
     /// JWT "Issued At" claim of JRL
     pub iat: i64,

devolutions-gateway/src/http/controllers/sessions.rs

@@ -31,8 +31,7 @@ impl SessionsController {
     security(("scope_token" = ["gateway.sessions.read"])),
 ))]
 pub(crate) async fn get_sessions() -> Json<Vec<GatewaySessionInfo>> {
-    let sessions = SESSIONS_IN_PROGRESS.read().await;
-    let sessions_in_progress: Vec<GatewaySessionInfo> = sessions.values().cloned().collect();
+    let sessions_in_progress: Vec<GatewaySessionInfo> = SESSIONS_IN_PROGRESS.read().values().cloned().collect();
     Json(sessions_in_progress)
 }