appcore_gateway/
service.rs

1// =============================================================================
2//        #######
3//     ###       ###     F: service.rs
4//    ##   ## ##   ##    P: AppCore-Runtime
5//         ## ##
6//                       C: 2026/07/26 08:53:09 by dnettoRaw
7//    ##   ## ##   ##    U: 2026/08/02 12:48:56 by dnettoRaw
8//      ###########      S: 1.0.1-rc.8
9// =============================================================================
10
11//! Axum HTTP and WebSocket routing service implementation.
12
13use crate::authorization::{
14    authenticate_connection, authenticate_mesh_request, client_connection_hash,
15    worker_connection_hash,
16};
17use crate::config::{
18    MAX_GATEWAY_CAPABILITIES, MAX_GATEWAY_HTTP_BODY_BYTES, MAX_GATEWAY_MESSAGE_BYTES,
19};
20use crate::mesh::{MeshPeerRequest, MESH_PEER_RELAY_PATH};
21use crate::socket::{handle_client_socket, handle_worker_socket, WorkerSocketContext};
22use crate::{EnvelopeRouter, GatewayState};
23use appcore_contracts::InstallationId;
24use appcore_security::RuntimeTokenClaims;
25use appcore_types::{CapabilityName, ClusterId, CoreId, InstanceId, TenantId};
26use axum::extract::{DefaultBodyLimit, Query, State, WebSocketUpgrade};
27use axum::http::{HeaderMap, StatusCode};
28use axum::response::{IntoResponse, Response};
29use axum::routing::get;
30use axum::{Json, Router};
31use serde::Deserialize;
32use std::collections::HashSet;
33use std::sync::Arc;
34use std::time::{Duration, SystemTime, UNIX_EPOCH};
35
36/// Connection parameters passed in the query string.
37#[derive(Deserialize)]
38pub struct ConnectionParams {
39    pub(crate) tenant: Option<String>,
40    pub(crate) cluster: Option<String>,
41    pub(crate) installation: Option<String>,
42    pub(crate) core: Option<String>,
43    pub(crate) device: Option<String>,
44    pub(crate) token: Option<String>,
45    pub(crate) capabilities: Option<String>,
46}
47
48impl std::fmt::Debug for ConnectionParams {
49    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
50        formatter
51            .debug_struct("ConnectionParams")
52            .field("tenant", &self.tenant)
53            .field("cluster", &self.cluster)
54            .field("installation", &self.installation)
55            .field("core", &self.core)
56            .field("device", &self.device)
57            .field("token", &self.token.as_ref().map(|_| "REDACTED"))
58            .field("capabilities", &self.capabilities)
59            .finish()
60    }
61}
62
63/// Resolves a tenant from the deployment domain, with a query fallback for test helpers.
64pub fn resolve_tenant(
65    headers: &HeaderMap,
66    params: &ConnectionParams,
67    domain_suffix: &str,
68) -> Option<TenantId> {
69    resolve_host_tenant(headers, domain_suffix)
70        .or_else(|| params.tenant.as_deref().and_then(valid_tenant))
71}
72
73/// Constructs the Axum router for the Gateway capability.
74pub fn make_gateway_router(state: Arc<GatewayState>) -> Router {
75    Router::new()
76        .route("/v1/gateway/worker/connect", get(worker_connect_handler))
77        .route("/v1/gateway/client/connect", get(client_connect_handler))
78        .route(
79            MESH_PEER_RELAY_PATH,
80            axum::routing::post(mesh_peer_relay_handler),
81        )
82        .layer(DefaultBodyLimit::max(MAX_GATEWAY_HTTP_BODY_BYTES))
83        .with_state(state)
84}
85
86async fn mesh_peer_relay_handler(
87    State(state): State<Arc<GatewayState>>,
88    headers: HeaderMap,
89    Json(request): Json<MeshPeerRequest>,
90) -> Response {
91    if state.is_shutting_down() {
92        return (StatusCode::SERVICE_UNAVAILABLE, "Gateway is shutting down").into_response();
93    }
94    if request.validate_schema().is_err() {
95        return (StatusCode::BAD_REQUEST, "Invalid mesh request").into_response();
96    }
97    if state.config().requires_authentication() {
98        let Some(token) = extract_token(&headers) else {
99            return (StatusCode::UNAUTHORIZED, "Missing credentials").into_response();
100        };
101        if authenticate_mesh_request(&state, token, &request, now_ms()).is_err() {
102            return (StatusCode::FORBIDDEN, "Invalid credentials").into_response();
103        }
104    }
105    let timeout = Duration::from_millis(request.timeout_ms);
106    let response = EnvelopeRouter::route_mesh_request(state, request, timeout).await;
107    (StatusCode::OK, Json(response)).into_response()
108}
109
110async fn worker_connect_handler(
111    ws: WebSocketUpgrade,
112    State(state): State<Arc<GatewayState>>,
113    headers: HeaderMap,
114    Query(params): Query<ConnectionParams>,
115) -> Response {
116    if state.is_shutting_down() {
117        return (StatusCode::SERVICE_UNAVAILABLE, "Gateway is shutting down").into_response();
118    }
119    let Some(tenant_id) = resolve_request_tenant(&state, &headers, &params) else {
120        return (StatusCode::BAD_REQUEST, "Missing or invalid tenant").into_response();
121    };
122    let Some(cluster_id) = params.cluster.as_ref().and_then(valid_cluster) else {
123        return (StatusCode::BAD_REQUEST, "Missing or invalid cluster").into_response();
124    };
125    let Some(installation_id) = params
126        .installation
127        .as_ref()
128        .and_then(|value| InstallationId::new(value).ok())
129    else {
130        return (StatusCode::BAD_REQUEST, "Missing or invalid installation").into_response();
131    };
132    let Some(core_id) = params.core.as_ref().and_then(valid_core) else {
133        return (StatusCode::BAD_REQUEST, "Missing or invalid core").into_response();
134    };
135    let capabilities = match parse_capabilities(params.capabilities.as_deref()) {
136        Ok(capabilities) => capabilities,
137        Err(message) => return (StatusCode::BAD_REQUEST, message).into_response(),
138    };
139    let expected_hash = worker_connection_hash(
140        &tenant_id,
141        &cluster_id,
142        &installation_id,
143        &core_id,
144        &capabilities,
145    );
146    let claims = match authenticate_upgrade(&state, &headers, &params, &expected_hash) {
147        Ok(claims) => claims,
148        Err(error) => return error.into_response(),
149    };
150    ws.max_message_size(MAX_GATEWAY_MESSAGE_BYTES)
151        .max_frame_size(MAX_GATEWAY_MESSAGE_BYTES)
152        .on_upgrade(move |socket| {
153            handle_worker_socket(
154                state,
155                WorkerSocketContext {
156                    tenant_id,
157                    cluster_id,
158                    installation_id,
159                    core_id,
160                    capabilities,
161                    expires_at_ms: claims.expires_at_ms,
162                },
163                socket,
164            )
165        })
166}
167
168async fn client_connect_handler(
169    ws: WebSocketUpgrade,
170    State(state): State<Arc<GatewayState>>,
171    headers: HeaderMap,
172    Query(params): Query<ConnectionParams>,
173) -> Response {
174    if state.is_shutting_down() {
175        return (StatusCode::SERVICE_UNAVAILABLE, "Gateway is shutting down").into_response();
176    }
177    let Some(tenant_id) = resolve_request_tenant(&state, &headers, &params) else {
178        return (StatusCode::BAD_REQUEST, "Missing or invalid tenant").into_response();
179    };
180    let Some(cluster_id) = params.cluster.as_ref().and_then(valid_cluster) else {
181        return (StatusCode::BAD_REQUEST, "Missing or invalid cluster").into_response();
182    };
183    let Some(device_id) = params.device.as_ref().and_then(valid_device) else {
184        return (StatusCode::BAD_REQUEST, "Missing or invalid device").into_response();
185    };
186    let expected_hash = client_connection_hash(&tenant_id, &cluster_id, &device_id);
187    let claims = match authenticate_upgrade(&state, &headers, &params, &expected_hash) {
188        Ok(claims) => claims,
189        Err(error) => return error.into_response(),
190    };
191    ws.max_message_size(MAX_GATEWAY_MESSAGE_BYTES)
192        .max_frame_size(MAX_GATEWAY_MESSAGE_BYTES)
193        .on_upgrade(move |socket| {
194            handle_client_socket(state, tenant_id, cluster_id, claims, socket)
195        })
196}
197
198fn authenticate_upgrade(
199    state: &GatewayState,
200    headers: &HeaderMap,
201    params: &ConnectionParams,
202    expected_hash: &str,
203) -> Result<RuntimeTokenClaims, UpgradeError> {
204    if params.token.is_some() {
205        return Err(UpgradeError::QueryNotAllowed);
206    }
207    if !state.config().requires_authentication() {
208        return Ok(insecure_claims());
209    }
210    let token = extract_token(headers).ok_or(UpgradeError::Missing)?;
211    authenticate_connection(state, token, expected_hash, now_ms())
212        .map_err(|_| UpgradeError::Invalid)
213}
214
215fn resolve_request_tenant(
216    state: &GatewayState,
217    headers: &HeaderMap,
218    params: &ConnectionParams,
219) -> Option<TenantId> {
220    resolve_host_tenant(headers, &state.config().domain_suffix).or_else(|| {
221        state
222            .config()
223            .bind_address
224            .ip()
225            .is_loopback()
226            .then(|| params.tenant.as_deref().and_then(valid_tenant))
227            .flatten()
228    })
229}
230
231fn resolve_host_tenant(headers: &HeaderMap, domain_suffix: &str) -> Option<TenantId> {
232    let host = headers.get("host")?.to_str().ok()?;
233    let host = host.split(':').next().unwrap_or(host);
234    let prefix = host.strip_suffix(domain_suffix)?.strip_suffix('.')?;
235    valid_tenant(prefix)
236}
237
238fn parse_capabilities(value: Option<&str>) -> Result<Vec<CapabilityName>, &'static str> {
239    let Some(value) = value else {
240        return Ok(Vec::new());
241    };
242    let values = value.split(',').collect::<Vec<_>>();
243    if values.len() > MAX_GATEWAY_CAPABILITIES {
244        return Err("Too many capabilities");
245    }
246    let mut unique = HashSet::new();
247    let mut capabilities = Vec::with_capacity(values.len());
248    for value in values {
249        let capability = CapabilityName::new(value.trim()).map_err(|_| "Invalid capability")?;
250        if unique.insert(capability.clone()) {
251            capabilities.push(capability);
252        }
253    }
254    Ok(capabilities)
255}
256
257enum UpgradeError {
258    QueryNotAllowed,
259    Missing,
260    Invalid,
261}
262
263impl UpgradeError {
264    fn into_response(self) -> Response {
265        match self {
266            Self::QueryNotAllowed => (
267                StatusCode::BAD_REQUEST,
268                "Query credentials are not accepted",
269            )
270                .into_response(),
271            Self::Missing => (StatusCode::UNAUTHORIZED, "Missing credentials").into_response(),
272            Self::Invalid => (StatusCode::FORBIDDEN, "Invalid credentials").into_response(),
273        }
274    }
275}
276
277fn extract_token(headers: &HeaderMap) -> Option<&str> {
278    let value = headers.get("authorization")?.to_str().ok()?;
279    value
280        .strip_prefix("Bearer ")
281        .or_else(|| value.strip_prefix("bearer "))
282}
283
284fn valid_tenant(value: &str) -> Option<TenantId> {
285    TenantId::new(value).ok()
286}
287
288fn valid_cluster(value: &String) -> Option<ClusterId> {
289    ClusterId::new(value).ok()
290}
291
292fn valid_core(value: &String) -> Option<CoreId> {
293    CoreId::new(value).ok()
294}
295
296fn valid_device(value: &String) -> Option<InstanceId> {
297    InstanceId::new(value).ok()
298}
299
300fn now_ms() -> u64 {
301    SystemTime::now()
302        .duration_since(UNIX_EPOCH)
303        .map(|duration| duration.as_millis() as u64)
304        .unwrap_or(0)
305}
306
307fn insecure_claims() -> RuntimeTokenClaims {
308    RuntimeTokenClaims {
309        version: "v1".to_string(),
310        purpose: "peer".to_string(),
311        command_name: None,
312        scope: None,
313        subject: None,
314        issued_at_ms: now_ms(),
315        expires_at_ms: u64::MAX,
316        jti: None,
317        request_hash: None,
318    }
319}
appcore_gateway/service.rs

appcore_gateway/
service.rs
