From dc5417334b5cba9d3edc8f2bf20b4fc2c22fa4e8 Mon Sep 17 00:00:00 2001 From: Sunil Thakare Date: Thu, 23 Jul 2026 15:17:30 +0530 Subject: [PATCH] feat: harden runtime lifecycle and application credentials - enforce deterministic runtime lifecycle state transitions - add live graceful-to-forced shutdown escalation - align HTTP draining and worker shutdown with global deadline - guarantee deterministic shutdown hook ordering - add secure application client IDs and one-time client secrets - hash application secrets with BLAKE3 and constant-time verification - make credential creation and rotation transactionally auditable - enforce strict client_id authentication and redirect URI validation - add SQLite and PostgreSQL credential migrations - add application credential and runtime lifecycle acceptance tests - update Dioxus application management workflows - update security and architecture documentation --- Cargo.lock | 2 + Cargo.toml | 2 + README.md | 135 +++- docs/SECURITY.md | 15 +- src/api/applications.rs | 114 ++- src/api/router.rs | 4 + .../0017_update_applications_credentials.sql | 23 + .../0018_harden_application_credentials.sql | 10 + .../0017_update_applications_credentials.sql | 21 + .../0018_harden_application_credentials.sql | 7 + src/db/models/application.rs | 42 + src/db/repository/postgres/applications.rs | 142 +++- src/db/repository/sqlite/applications.rs | 152 +++- src/db/repository/traits.rs | 21 +- src/identity/applications.rs | 382 ++++++++- src/runtime/application.rs | 133 ++-- src/runtime/cancellation.rs | 54 +- src/runtime/hooks.rs | 7 +- src/runtime/signals.rs | 34 + src/runtime/state.rs | 88 ++- src/runtime/workers.rs | 99 ++- tests/application_credentials_test.rs | 722 ++++++++++++++++++ tests/runtime_lifecycle_test.rs | 233 +++++- ui/src/models/mod.rs | 15 + ui/src/pages/applications/mod.rs | 145 +++- ui/src/services/api.rs | 58 +- 26 files changed, 2477 insertions(+), 183 deletions(-) create mode 100644 src/db/migrations/postgres/0017_update_applications_credentials.sql create mode 100644 src/db/migrations/postgres/0018_harden_application_credentials.sql create mode 100644 src/db/migrations/sqlite/0017_update_applications_credentials.sql create mode 100644 src/db/migrations/sqlite/0018_harden_application_credentials.sql create mode 100644 tests/application_credentials_test.rs diff --git a/Cargo.lock b/Cargo.lock index 909c4c4..a8940be 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1265,6 +1265,7 @@ dependencies = [ "serde", "serde_json", "sqlx", + "subtle", "thiserror", "time", "tokio", @@ -1274,6 +1275,7 @@ dependencies = [ "tower-http", "tracing", "tracing-subscriber", + "url", "uuid", ] diff --git a/Cargo.toml b/Cargo.toml index c2341e9..10db16f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -73,6 +73,8 @@ dashmap = "6.0" # Utilities hex = "0.4" +url = "2.5" +subtle = "2.6" [profile.release] opt-level = 3 diff --git a/README.md b/README.md index 07fe4cc..8e87c4b 100644 --- a/README.md +++ b/README.md @@ -4,7 +4,7 @@ **Enterprise Identity & Access Management (IAM)** -*Self-Hosted • Privacy-First • Pure Rust • Single Binary • Dual Database Engine* +*Self-Hosted • Privacy-First • Pure Rust • Single Binary • SQLite & PostgreSQL* [![Version](https://img.shields.io/badge/version-v0.3.0-blue.svg)]() [![Rust](https://img.shields.io/badge/Rust-2024-orange.svg)](https://www.rust-lang.org/) @@ -19,21 +19,70 @@ ## Overview -**nx9-auth** is a production-grade, self-hosted Identity & Access Management (IAM) server built entirely in **Rust**. It provides multi-tenant user authentication, Role-Based Access Control (RBAC), Personal Access Tokens (PATs), OAuth2 service accounts, active session management, full audit logging, an enterprise graceful shutdown runtime lifecycle, and an embedded WebAssembly (WASM) administrative UI. +**nx9-auth** is a self-hosted Identity & Access Management (IAM) server built entirely in **Rust**. It provides centralized authentication, multi-tenancy, fine-grained Role-Based Access Control (RBAC), Personal Access Tokens (PATs), service accounts, application registration credentials, active session management, and append-only audit logging. -`nx9-auth` compiles into a single standalone binary containing both the Axum REST API backend and the embedded Dioxus WASM frontend, backed by a database-agnostic provider supporting both **SQLite** and **PostgreSQL**. +The server uses **Axum**, **Tokio**, and **SQLx**, with repository implementations for both embedded **SQLite** and external **PostgreSQL** deployments. Its administration interface is built with **Dioxus 0.6** and compiled to **WebAssembly (WASM)**, requiring no Node.js or npm runtime/build chain for the application architecture. + +The runtime lifecycle and Application Registration Credentials subsystems have completed dedicated production-hardening passes covering deterministic shutdown, live signal escalation, transactional credential operations, secret handling, authorization boundaries, redirect URI validation, and regression testing. --- ## Key Features -- **Unified Enterprise Runtime Lifecycle**: Atomic 8-state lifecycle machine (`Initializing` → `Starting` → `Running` → `Draining` → `StoppingWorkers` → `ExecutingHooks` → `ClosingResources` → `Stopped`), `CancellationToken` propagation, `JoinSet` worker management, prioritized shutdown hooks, and destructor-safe Unix signal escalation. -- **Dual Database Engine**: Native support for SQLite and enterprise PostgreSQL with 100% repository parity and runtime connection pool ownership. -- **Enterprise Security Model**: Argon2id password hashing, BLAKE3 token/session hashing, rate-limiting, CSP, HSTS, and non-enumerating authentication. -- **Multi-Tenant & RBAC**: Tenant isolation, fine-grained permission matrix, role assignments, and organizational user groups. -- **Personal Access Tokens & Service Accounts**: Machine-to-machine authentication with automatic prefix tracking and instant revocation. -- **Embedded WebAssembly UI**: Dioxus-powered administration dashboard with `#boot-loader` lifecycle management. -- **Comprehensive CLI Tooling**: Automated `init`, `doctor`, `migrate`, `backup`, `restore`, and user management commands. +- **Deterministic Runtime Lifecycle**: Atomic 8-state lifecycle (`Initializing` → `Starting` → `Running` → `Draining` → `StoppingWorkers` → `ExecutingHooks` → `ClosingResources` → `Stopped`) with graph-validated transitions, cancellation propagation, supervised worker shutdown, HTTP draining, prioritized shutdown hooks, and live forced escalation on a second Unix termination signal. +- **SQLite & PostgreSQL Support**: Shared repository abstraction with backend-specific migrations and repository implementations for embedded SQLite and external PostgreSQL deployments. +- **Security-Oriented Authentication**: Argon2id password hashing, BLAKE3 credential/token digests, constant-time credential comparison, rate limiting, non-enumerating authentication failures, secret redaction, and security response headers. +- **Multi-Tenant RBAC**: Tenant-aware identities, fine-grained permissions, role assignments, and organizational user groups. +- **Personal Access Tokens & Service Accounts**: Credentials for API and machine-to-machine access with hashed-at-rest secrets and revocation support. +- **Application Registration Credentials**: Immutable server-generated Client IDs, one-time Client Secret disclosure, BLAKE3 secret hashing, constant-time verification, secret rotation, redirect URI metadata, scopes, and dedicated `applications:manage` authorization. +- **Transactional Credential Integrity**: Application creation and Client Secret rotation are committed atomically with their audit records; audit failure rolls back the associated credential operation. +- **Strict Application Identity**: Application authentication uses `client_id` only; editable application slugs are never accepted as credential identities. +- **Redirect URI Policy**: Registered redirect URIs are structurally validated. HTTPS is supported generally; HTTP is restricted to localhost/loopback development destinations. Fragments, userinfo credentials, unsupported schemes, excessive URI counts, and oversized entries are rejected. +- **Embedded WebAssembly Administration UI**: Dioxus-powered administration interface compiled to WASM without a Node.js/React frontend stack. +- **Structured Auditability**: Security-sensitive lifecycle and identity operations are audit logged while plaintext passwords, Client Secrets, tokens, and stored credential hashes are excluded from audit metadata. +- **CLI Tooling**: Command-line workflows for initialization, diagnostics, migration, backup/restore, server operation, and identity administration. + +--- + +## Application Registration Credentials + +Applications are registered with a stable public identity and a high-entropy secret: + +```text +Client ID: nx9_app_<32 lowercase hex characters> +Client Secret: nx9_secret_<64 lowercase hex characters> +``` + +The **Client ID** is immutable and safe to identify an application. The **Client Secret** is disclosed only when the application is created or its secret is explicitly rotated. + +Plaintext Client Secrets are never persisted. NX9-Auth stores a BLAKE3 digest and performs credential comparison using constant-time byte comparison. Creation and secret rotation responses are treated as one-time secret disclosure operations and use `Cache-Control: no-store`. + +Existing applications upgraded from earlier schemas receive a stable Client ID. Applications without previously configured credentials can establish credentials through explicit secret rotation. + +> **Protocol boundary:** Application credentials, redirect URIs, and scopes form the application registration layer. Redirect URIs are registration metadata intended to become security-enforced redirect destinations when OAuth2/OIDC protocol handlers are implemented. This registration subsystem does not by itself claim complete OAuth2/OIDC grant-flow support. + +--- + +## Runtime Lifecycle + +NX9-Auth uses an explicit lifecycle graph: + +```mermaid +stateDiagram-v2 + [*] --> Initializing + Initializing --> Starting: Build Runtime + Starting --> Running: Start Services + Running --> Draining: Begin Shutdown + Draining --> StoppingWorkers: HTTP Drain Completes or Is Force-Aborted + StoppingWorkers --> ExecutingHooks: Workers Terminated + ExecutingHooks --> ClosingResources: Hooks Complete + ClosingResources --> Stopped: Resources Closed + Stopped --> [*] +``` + +The first `SIGINT` or `SIGTERM` initiates graceful shutdown, transitions the runtime into `Draining`, and begins HTTP request draining. Signal monitoring remains active throughout shutdown. A second termination signal escalates shutdown immediately, allowing HTTP draining and blocked worker waits to be curtailed rather than consuming the remaining graceful deadline. + +Worker groups receive cancellation concurrently and operate under a shared global shutdown budget. Shutdown hooks execute deterministically by priority, with hooks at the same priority retaining registration order. --- @@ -43,7 +92,7 @@ # Initialize application directory, configuration, and default administrator nx9-auth init -# Verify installation & system health +# Verify installation and system health nx9-auth doctor # Start server @@ -54,7 +103,7 @@ nx9-auth serve ## Configuration -Configure `config.toml` or set environment variables: +Configure `config.toml` or use the supported environment-variable configuration: ```toml [server] @@ -67,7 +116,7 @@ cookie_secure = false # SQLite URL or file path: url = "sqlite://./data/auth.db?mode=rwc" -# Or enterprise PostgreSQL: +# Or PostgreSQL: # url = "postgres://user:password@localhost:5432/nx9auth" max_connections = 20 @@ -81,11 +130,56 @@ graceful_timeout_secs = 30 force_timeout_secs = 35 ``` +For production deployments, terminate TLS appropriately, use secure cookies, protect configuration and database credentials, and apply deployment-specific filesystem and network permissions. + +--- + +## Security Model + +NX9-Auth applies layered controls rather than relying on any single authentication mechanism: + +| Area | Control | +|---|---| +| Passwords | Argon2id password hashing | +| Application secrets | BLAKE3 digest at rest | +| Credential comparison | `subtle::ConstantTimeEq` | +| Authentication failures | Non-enumerating unauthorized responses | +| Application mutations | Dedicated `applications:manage` permission | +| Application creation | Transactional application + audit insertion | +| Secret rotation | Transactional secret update + audit insertion | +| Secret disclosure | One-time response; never returned by list/GET operations | +| Redirect URIs | Structural and scheme-policy validation | +| HTTP responses | Security headers and no-store handling for secret responses | +| Audit metadata | Secret and credential-hash redaction | + +Security controls documented here describe implemented mechanisms and should not be interpreted as a substitute for deployment-specific threat modelling, security review, or external audit. + +--- + +## Verification + +The runtime lifecycle and Application Registration Credentials hardening scopes are covered by workspace unit, integration, migration, security, and acceptance tests. + +The verification gates used for these scopes are: + +```bash +cargo fmt --all -- --check +cargo check --workspace --all-targets --all-features +cargo clippy --workspace --all-targets --all-features -- -D warnings +cargo test --workspace --all-features +cargo build --release +cargo check --manifest-path ui/Cargo.toml --target wasm32-unknown-unknown +``` + +At the documented hardening checkpoint, the workspace test suite completed with **97 tests passed and 0 failed**. Test counts and execution times are verification-run observations rather than performance guarantees. + --- ## Documentation Index -- [Runtime Lifecycle & Graceful Shutdown](docs/runtime-lifecycle.md) +- [System Architecture](docs/ARCHITECTURE.md) +- [Runtime Lifecycle & Graceful Shutdown](docs/RUNTIME_LIFECYCLE.md) +- [Security Architecture](docs/SECURITY.md) - [Release Notes](RELEASE_NOTES.md) - [Authentication Model](docs/AUTHENTICATION.md) - [Backup & Disaster Recovery](docs/BACKUPS.md) @@ -98,10 +192,19 @@ force_timeout_secs = 35 --- +## Project Status + +The **Runtime Lifecycle** and **Application Registration Credentials** hardening scopes documented for the current release are complete. + +Future OAuth2/OIDC protocol handlers, additional authentication protocols, deployment hardening, or architectural changes should be introduced as separately scoped work with corresponding migrations, security review, and regression tests. + +--- + ## License Dual-licensed under either of: -- Apache License, Version 2.0 ([LICENSE](LICENSE) or http://www.apache.org/licenses/LICENSE-2.0) -- MIT License ([LICENSE](LICENSE) or http://opensource.org/licenses/MIT) + +- Apache License, Version 2.0 ([LICENSE](LICENSE) or ) +- MIT License ([LICENSE](LICENSE) or ) at your option. diff --git a/docs/SECURITY.md b/docs/SECURITY.md index 3d99f4a..aa4df5e 100644 --- a/docs/SECURITY.md +++ b/docs/SECURITY.md @@ -22,12 +22,21 @@ NX9-Auth is designed with a **security-first, privacy-first, zero-trust** archit - `Permissions-Policy: accelerometer=(), camera=(), geolocation=(), ...` - `Strict-Transport-Security: max-age=63072000; includeSubDomains` (when `cookie_secure` / production is enabled) +## Application Credentials & Client Authentication + +- **Client ID & Client Secret**: Applications registered in NX9-Auth receive an immutable, server-generated `client_id` (`nx9_app_<32 lowercase hex chars>`) and high-entropy CSPRNG `client_secret` (`nx9_secret_<64 lowercase hex chars>`). +- **One-Time Secret Disclosure**: Plaintext client secrets are disclosed **exactly once** upon initial application creation and explicit secret rotation. Responses containing plaintext secrets include `Cache-Control: no-store` headers. +- **BLAKE3 Secret Hashing**: Only BLAKE3 cryptographic digests (`[u8; 32]`) are persisted in database records. Plaintext secrets are never stored, logged, serialized into GET/list API responses, or stored in browser persistence. +- **Constant-Time Raw Byte Verification**: Verification hashes supplied credentials to a 32-byte BLAKE3 digest and constant-time compares bytes against the stored 32-byte digest. To prevent timing side-channel attacks for unknown or uncredentialed applications, a dummy BLAKE3 comparison path is executed before returning non-enumerating `401 Unauthorized` errors. +- **Secret Rotation**: Administrator rotation immediately invalidates the previous client secret hash and generates a new secret. +- **Dedicated Permissions**: Application mutations (`create`, `update`, `rotate_secret`, `enable_disable`, `delete`) require the `applications:manage` permission. + ## Audit Logging Security Audit logs record critical identity lifecycle events while strictly redacting sensitive fields: -- **Recorded Events**: Login success/failure, logout, password change, user creation/deletion, API token issuance/revocation, role/permission assignments. -- **Redaction Rules**: Plaintext passwords, password hashes, bearer tokens, refresh tokens, session secrets, and `Authorization` headers are **never** logged under any circumstances. +- **Recorded Events**: Login success/failure, logout, password change, user creation/deletion, API token issuance/revocation, application creation/secret rotation/modification, role/permission assignments. +- **Redaction Rules**: Plaintext passwords, password hashes, bearer tokens, refresh tokens, client secrets, client secret hashes, session secrets, and `Authorization` headers are **never** logged under any circumstances. ## Rate Limiting & Protection -- **Progressive Lockout**: Progressive rate limiting protects sensitive endpoints (`/auth/login`, `/users/{id}/reset-password`, `/tokens`) against brute-force and credential-stuffing attacks. +- **Progressive Lockout**: Progressive rate limiting protects sensitive endpoints (`/auth/login`, `/users/{id}/reset-password`, `/tokens`, `/applications/{id}/secret`) against brute-force and credential-stuffing attacks. diff --git a/src/api/applications.rs b/src/api/applications.rs index 563a3a6..89b7332 100644 --- a/src/api/applications.rs +++ b/src/api/applications.rs @@ -1,6 +1,7 @@ use axum::{ Json, extract::{Path, State}, + http::{HeaderMap, HeaderValue, header}, }; use serde::{Deserialize, Serialize}; use serde_json::{Value, json}; @@ -13,14 +14,17 @@ use crate::{ state::AppState, }; +pub const MANAGE_PERM: &str = "applications:manage"; + #[derive(Serialize)] pub struct ApplicationResponse { pub id: String, pub name: String, pub slug: String, - /// Client ID — currently the application slug (OAuth2-ready). pub client_id: String, + pub description: Option, pub enabled: bool, + pub credentials_configured: bool, pub redirect_urls: Vec, pub scopes: Vec, pub created_at: String, @@ -29,30 +33,51 @@ pub struct ApplicationResponse { impl From for ApplicationResponse { fn from(a: Application) -> Self { + let client_id = a.get_client_id().to_string(); + let redirect_urls = a.redirect_urls(); + let scopes = a.scopes(); + let credentials_configured = a.has_credentials(); Self { id: a.id, name: a.name, - client_id: a.slug.clone().unwrap_or_default(), slug: a.slug.unwrap_or_default(), + client_id, + description: a.description, enabled: a.enabled, - // Placeholder until OAuth2 tables land - redirect_urls: Vec::new(), - scopes: Vec::new(), + credentials_configured, + redirect_urls, + scopes, created_at: a.created_at, updated_at: a.updated_at, } } } +#[derive(Serialize)] +pub struct CreateApplicationResponse { + pub application: ApplicationResponse, + pub client_secret: String, +} + +#[derive(Serialize)] +pub struct RotateSecretResponse { + pub client_secret: String, +} + +fn no_store_headers() -> HeaderMap { + let mut headers = HeaderMap::new(); + headers.insert(header::CACHE_CONTROL, HeaderValue::from_static("no-store")); + headers +} + /// GET /api/v1/applications pub async fn list_applications( State(state): State, auth: AuthUser, ) -> Result> { - // Any authenticated user can see registered apps; mutations need roles:manage + let _ = auth; let apps = identity::list(&state.provider, Tenant::DEFAULT_ID).await?; let views: Vec = apps.into_iter().map(ApplicationResponse::from).collect(); - let _ = auth; Ok(Json(json!({ "applications": views }))) } @@ -60,6 +85,9 @@ pub async fn list_applications( pub struct CreateApplicationRequest { pub name: String, pub slug: String, + pub description: Option, + pub redirect_urls: Option>, + pub scopes: Option>, } /// POST /api/v1/applications @@ -67,13 +95,29 @@ pub async fn create_application( State(state): State, auth: AuthUser, Json(body): Json, -) -> Result> { - require(&state.provider, &auth.user.id, "roles:manage").await?; +) -> Result<(HeaderMap, Json)> { + require(&state.provider, &auth.user.id, MANAGE_PERM).await?; - let app = identity::create(&state.provider, Tenant::DEFAULT_ID, &body.name, &body.slug).await?; - Ok(Json( - json!({ "application": ApplicationResponse::from(app) }), - )) + let (app, raw_secret) = identity::create( + &state.provider, + Tenant::DEFAULT_ID, + &body.name, + &body.slug, + body.description.as_deref(), + body.redirect_urls, + body.scopes, + Some(&auth.user.id), + None, + None, + ) + .await?; + + let resp = CreateApplicationResponse { + application: ApplicationResponse::from(app), + client_secret: raw_secret, + }; + + Ok((no_store_headers(), Json(resp))) } /// GET /api/v1/applications/:id @@ -90,9 +134,13 @@ pub async fn get_application( } #[derive(Debug, Deserialize)] +#[serde(deny_unknown_fields)] pub struct UpdateApplicationRequest { pub name: String, pub slug: String, + pub description: Option, + pub redirect_urls: Option>, + pub scopes: Option>, pub enabled: bool, } @@ -103,21 +151,53 @@ pub async fn update_application( Path(id): Path, Json(body): Json, ) -> Result> { - require(&state.provider, &auth.user.id, "roles:manage").await?; + require(&state.provider, &auth.user.id, MANAGE_PERM).await?; + + let app = identity::update( + &state.provider, + &id, + &body.name, + &body.slug, + body.description.as_deref(), + body.redirect_urls, + body.scopes, + body.enabled, + Some(&auth.user.id), + None, + None, + ) + .await?; - let app = identity::update(&state.provider, &id, &body.name, &body.slug, body.enabled).await?; Ok(Json( json!({ "application": ApplicationResponse::from(app) }), )) } +/// POST /api/v1/applications/:id/secret +pub async fn rotate_application_secret( + State(state): State, + auth: AuthUser, + Path(id): Path, +) -> Result<(HeaderMap, Json)> { + require(&state.provider, &auth.user.id, MANAGE_PERM).await?; + + let raw_secret = + identity::rotate_secret(&state.provider, &id, Some(&auth.user.id), None, None).await?; + + let resp = RotateSecretResponse { + client_secret: raw_secret, + }; + + Ok((no_store_headers(), Json(resp))) +} + /// DELETE /api/v1/applications/:id pub async fn delete_application( State(state): State, auth: AuthUser, Path(id): Path, ) -> Result> { - require(&state.provider, &auth.user.id, "roles:manage").await?; - identity::delete(&state.provider, &id).await?; + require(&state.provider, &auth.user.id, MANAGE_PERM).await?; + identity::delete(&state.provider, &id, Some(&auth.user.id), None, None).await?; Ok(Json(json!({ "success": true }))) } diff --git a/src/api/router.rs b/src/api/router.rs index 0c78250..f368092 100644 --- a/src/api/router.rs +++ b/src/api/router.rs @@ -82,6 +82,10 @@ pub fn build(state: AppState) -> Router { .patch(applications::update_application) .delete(applications::delete_application), ) + .route( + "/applications/{id}/secret", + post(applications::rotate_application_secret), + ) // Service accounts .route( "/service-accounts", diff --git a/src/db/migrations/postgres/0017_update_applications_credentials.sql b/src/db/migrations/postgres/0017_update_applications_credentials.sql new file mode 100644 index 0000000..6cfb8cd --- /dev/null +++ b/src/db/migrations/postgres/0017_update_applications_credentials.sql @@ -0,0 +1,23 @@ +-- ── Add Application Credentials Columns & Permissions (PostgreSQL) ─────────── + +ALTER TABLE applications ADD COLUMN IF NOT EXISTS client_id TEXT; +ALTER TABLE applications ADD COLUMN IF NOT EXISTS description TEXT; +ALTER TABLE applications ADD COLUMN IF NOT EXISTS client_secret_hash TEXT; +ALTER TABLE applications ADD COLUMN IF NOT EXISTS redirect_uris TEXT; +ALTER TABLE applications ADD COLUMN IF NOT EXISTS scopes TEXT; + +-- Backfill client_id for existing applications +UPDATE applications SET client_id = 'nx9_app_' || replace(id, '-', '') WHERE client_id IS NULL; + +-- Create unique index on client_id +CREATE UNIQUE INDEX IF NOT EXISTS idx_applications_client_id ON applications(client_id); + +-- Seed applications:manage permission +INSERT INTO permissions (id, name, description) VALUES + ('20000000-0000-0000-0000-000000000008', 'applications:manage', 'Manage registered application credentials') +ON CONFLICT (name) DO NOTHING; + +-- Grant permission to admin role +INSERT INTO role_permissions (role_id, permission_id) VALUES + ('10000000-0000-0000-0000-000000000001', '20000000-0000-0000-0000-000000000008') +ON CONFLICT DO NOTHING; diff --git a/src/db/migrations/postgres/0018_harden_application_credentials.sql b/src/db/migrations/postgres/0018_harden_application_credentials.sql new file mode 100644 index 0000000..d5da15d --- /dev/null +++ b/src/db/migrations/postgres/0018_harden_application_credentials.sql @@ -0,0 +1,10 @@ +-- ── Application Credentials Production Hardening (PostgreSQL) ─────────────── + +-- Backfill any remaining applications with client_id if missing +UPDATE applications SET client_id = 'nx9_app_' || replace(id, '-', '') WHERE client_id IS NULL; + +-- Enforce NOT NULL constraint on client_id +ALTER TABLE applications ALTER COLUMN client_id SET NOT NULL; + +-- Ensure unique index on client_id exists +CREATE UNIQUE INDEX IF NOT EXISTS idx_applications_client_id ON applications(client_id); diff --git a/src/db/migrations/sqlite/0017_update_applications_credentials.sql b/src/db/migrations/sqlite/0017_update_applications_credentials.sql new file mode 100644 index 0000000..740bc31 --- /dev/null +++ b/src/db/migrations/sqlite/0017_update_applications_credentials.sql @@ -0,0 +1,21 @@ +-- ── Add Application Credentials Columns & Permissions (SQLite) ──────────────── + +ALTER TABLE applications ADD COLUMN client_id TEXT; +ALTER TABLE applications ADD COLUMN description TEXT; +ALTER TABLE applications ADD COLUMN client_secret_hash TEXT; +ALTER TABLE applications ADD COLUMN redirect_uris TEXT; +ALTER TABLE applications ADD COLUMN scopes TEXT; + +-- Backfill client_id for existing applications +UPDATE applications SET client_id = 'nx9_app_' || replace(id, '-', '') WHERE client_id IS NULL; + +-- Create unique index on client_id +CREATE UNIQUE INDEX IF NOT EXISTS idx_applications_client_id ON applications(client_id); + +-- Seed applications:manage permission +INSERT OR IGNORE INTO permissions (id, name, description) VALUES + ('20000000-0000-0000-0000-000000000008', 'applications:manage', 'Manage registered application credentials'); + +-- Grant permission to admin role +INSERT OR IGNORE INTO role_permissions (role_id, permission_id) VALUES + ('10000000-0000-0000-0000-000000000001', '20000000-0000-0000-0000-000000000008'); diff --git a/src/db/migrations/sqlite/0018_harden_application_credentials.sql b/src/db/migrations/sqlite/0018_harden_application_credentials.sql new file mode 100644 index 0000000..b6cb5cd --- /dev/null +++ b/src/db/migrations/sqlite/0018_harden_application_credentials.sql @@ -0,0 +1,7 @@ +-- ── Application Credentials Production Hardening (SQLite) ─────────────────── + +-- Backfill any remaining applications with client_id if missing +UPDATE applications SET client_id = 'nx9_app_' || replace(id, '-', '') WHERE client_id IS NULL; + +-- Ensure unique index on client_id exists +CREATE UNIQUE INDEX IF NOT EXISTS idx_applications_client_id ON applications(client_id); diff --git a/src/db/models/application.rs b/src/db/models/application.rs index 266edd9..2c60ca3 100644 --- a/src/db/models/application.rs +++ b/src/db/models/application.rs @@ -8,9 +8,51 @@ pub struct Application { pub name: String, pub description: Option, pub slug: Option, + pub client_id: String, pub enabled: bool, pub client_secret_hash: Option, pub redirect_uris: Option, + pub scopes: Option, pub created_at: String, pub updated_at: String, } + +impl Application { + /// Return effective client ID string. + pub fn get_client_id(&self) -> &str { + &self.client_id + } + + /// Parse configured redirect URLs. + pub fn redirect_urls(&self) -> Vec { + let Some(raw) = &self.redirect_uris else { + return Vec::new(); + }; + if let Ok(vec) = serde_json::from_str::>(raw) { + return vec; + } + raw.split([',', '\n', ' ']) + .map(|s| s.trim().to_string()) + .filter(|s| !s.is_empty()) + .collect() + } + + /// Parse configured scopes. + pub fn scopes(&self) -> Vec { + let Some(raw) = &self.scopes else { + return Vec::new(); + }; + if let Ok(vec) = serde_json::from_str::>(raw) { + return vec; + } + raw.split([',', ' ']) + .map(|s| s.trim().to_string()) + .filter(|s| !s.is_empty()) + .collect() + } + + /// Return true if application has configured client secret credentials. + pub fn has_credentials(&self) -> bool { + self.client_secret_hash.is_some() + } +} diff --git a/src/db/repository/postgres/applications.rs b/src/db/repository/postgres/applications.rs index 1660420..3cc9a33 100644 --- a/src/db/repository/postgres/applications.rs +++ b/src/db/repository/postgres/applications.rs @@ -10,17 +10,25 @@ pub struct PostgresApplicationsRepository { #[async_trait] impl ApplicationsRepository for PostgresApplicationsRepository { - async fn create( + async fn create_with_audit( &self, id: &str, tenant_id: &str, name: &str, slug: &str, + client_id: &str, + client_secret_hash: Option<&str>, + description: Option<&str>, + redirect_uris: Option<&str>, + scopes: Option<&str>, + audit_event: Option>, ) -> Result { - sqlx::query_as::<_, Application>( + let mut tx = self.pool.begin().await?; + + let app = sqlx::query_as::<_, Application>( r#" - INSERT INTO applications (id, tenant_id, name, slug) - VALUES ($1, $2, $3, $4) + INSERT INTO applications (id, tenant_id, name, slug, client_id, client_secret_hash, description, redirect_uris, scopes) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) RETURNING * "#, ) @@ -28,8 +36,43 @@ impl ApplicationsRepository for PostgresApplicationsRepository { .bind(tenant_id) .bind(name) .bind(slug) - .fetch_one(&self.pool) - .await + .bind(client_id) + .bind(client_secret_hash) + .bind(description) + .bind(redirect_uris) + .bind(scopes) + .fetch_one(&mut *tx) + .await?; + + if let Some(event) = audit_event { + let audit_id = uuid::Uuid::new_v4().to_string(); + let severity_str = event.severity.to_string(); + sqlx::query( + r#" + INSERT INTO audit_logs ( + id, actor_user_id, target_user_id, + action, resource_type, resource_id, + severity, ip_address, user_agent, metadata_json + ) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10) + "#, + ) + .bind(&audit_id) + .bind(event.actor_id) + .bind(event.target_id) + .bind(event.action) + .bind(event.resource_type) + .bind(event.resource_id) + .bind(&severity_str) + .bind(event.ip) + .bind(event.ua) + .bind(event.metadata) + .execute(&mut *tx) + .await?; + } + + tx.commit().await?; + Ok(app) } async fn find_by_slug(&self, slug: &str) -> Result, sqlx::Error> { @@ -39,6 +82,13 @@ impl ApplicationsRepository for PostgresApplicationsRepository { .await } + async fn find_by_client_id(&self, client_id: &str) -> Result, sqlx::Error> { + sqlx::query_as::<_, Application>("SELECT * FROM applications WHERE client_id = $1") + .bind(client_id) + .fetch_optional(&self.pool) + .await + } + async fn find_by_id(&self, id: &str) -> Result, sqlx::Error> { sqlx::query_as::<_, Application>("SELECT * FROM applications WHERE id = $1") .bind(id) @@ -59,10 +109,72 @@ impl ApplicationsRepository for PostgresApplicationsRepository { sqlx::query( "UPDATE applications SET enabled = $1, updated_at = to_char(clock_timestamp() AT TIME ZONE 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') WHERE id = $2", ) - .bind(enabled) - .bind(id) - .execute(&self.pool) - .await?; + .bind(enabled) + .bind(id) + .execute(&self.pool) + .await?; + Ok(()) + } + + async fn update_secret_hash(&self, id: &str, secret_hash: &str) -> Result<(), sqlx::Error> { + sqlx::query( + "UPDATE applications SET client_secret_hash = $1, updated_at = to_char(clock_timestamp() AT TIME ZONE 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') WHERE id = $2", + ) + .bind(secret_hash) + .bind(id) + .execute(&self.pool) + .await?; + Ok(()) + } + + async fn rotate_secret_with_audit( + &self, + id: &str, + secret_hash: &str, + audit_event: Option>, + ) -> Result<(), sqlx::Error> { + let mut tx = self.pool.begin().await?; + + let res = sqlx::query( + "UPDATE applications SET client_secret_hash = $1, updated_at = to_char(clock_timestamp() AT TIME ZONE 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') WHERE id = $2", + ) + .bind(secret_hash) + .bind(id) + .execute(&mut *tx) + .await?; + + if res.rows_affected() == 0 { + return Err(sqlx::Error::RowNotFound); + } + + if let Some(event) = audit_event { + let audit_id = uuid::Uuid::new_v4().to_string(); + let severity_str = event.severity.to_string(); + sqlx::query( + r#" + INSERT INTO audit_logs ( + id, actor_user_id, target_user_id, + action, resource_type, resource_id, + severity, ip_address, user_agent, metadata_json + ) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10) + "#, + ) + .bind(&audit_id) + .bind(event.actor_id) + .bind(event.target_id) + .bind(event.action) + .bind(event.resource_type) + .bind(event.resource_id) + .bind(&severity_str) + .bind(event.ip) + .bind(event.ua) + .bind(event.metadata) + .execute(&mut *tx) + .await?; + } + + tx.commit().await?; Ok(()) } @@ -71,18 +183,24 @@ impl ApplicationsRepository for PostgresApplicationsRepository { id: &str, name: &str, slug: &str, + description: Option<&str>, + redirect_uris: Option<&str>, + scopes: Option<&str>, enabled: bool, ) -> Result<(), sqlx::Error> { sqlx::query( r#" UPDATE applications - SET name = $1, slug = $2, enabled = $3, + SET name = $1, slug = $2, description = $3, redirect_uris = $4, scopes = $5, enabled = $6, updated_at = to_char(clock_timestamp() AT TIME ZONE 'UTC', 'YYYY-MM-DD"T"HH24:MI:SS"Z"') - WHERE id = $4 + WHERE id = $7 "#, ) .bind(name) .bind(slug) + .bind(description) + .bind(redirect_uris) + .bind(scopes) .bind(enabled) .bind(id) .execute(&self.pool) diff --git a/src/db/repository/sqlite/applications.rs b/src/db/repository/sqlite/applications.rs index dae4be6..9569c84 100644 --- a/src/db/repository/sqlite/applications.rs +++ b/src/db/repository/sqlite/applications.rs @@ -10,37 +10,87 @@ pub struct SqliteApplicationsRepository { #[async_trait] impl ApplicationsRepository for SqliteApplicationsRepository { - async fn create( + async fn create_with_audit( &self, id: &str, tenant_id: &str, name: &str, slug: &str, + client_id: &str, + client_secret_hash: Option<&str>, + description: Option<&str>, + redirect_uris: Option<&str>, + scopes: Option<&str>, + audit_event: Option>, ) -> Result { - sqlx::query_as::<_, Application>( + let mut tx = self.pool.begin().await?; + + let app = sqlx::query_as::<_, Application>( r#" - INSERT INTO applications (id, tenant_id, name, slug) - VALUES (?, ?, ?, ?) - RETURNING id, tenant_id, name, slug, enabled, created_at, updated_at, NULL as description, NULL as client_secret_hash, NULL as redirect_uris + INSERT INTO applications (id, tenant_id, name, slug, client_id, client_secret_hash, description, redirect_uris, scopes) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) + RETURNING id, tenant_id, name, slug, client_id, description, enabled, client_secret_hash, redirect_uris, scopes, created_at, updated_at "#, ) .bind(id) .bind(tenant_id) .bind(name) .bind(slug) - .fetch_one(&self.pool) - .await + .bind(client_id) + .bind(client_secret_hash) + .bind(description) + .bind(redirect_uris) + .bind(scopes) + .fetch_one(&mut *tx) + .await?; + + if let Some(event) = audit_event { + let audit_id = uuid::Uuid::new_v4().to_string(); + let severity_str = event.severity.to_string(); + sqlx::query( + r#" + INSERT INTO audit_logs ( + id, actor_user_id, target_user_id, + action, resource_type, resource_id, + severity, ip_address, user_agent, metadata_json + ) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + "#, + ) + .bind(&audit_id) + .bind(event.actor_id) + .bind(event.target_id) + .bind(event.action) + .bind(event.resource_type) + .bind(event.resource_id) + .bind(&severity_str) + .bind(event.ip) + .bind(event.ua) + .bind(event.metadata) + .execute(&mut *tx) + .await?; + } + + tx.commit().await?; + Ok(app) } async fn find_by_slug(&self, slug: &str) -> Result, sqlx::Error> { - sqlx::query_as::<_, Application>("SELECT id, tenant_id, name, slug, enabled, created_at, updated_at, NULL as description, NULL as client_secret_hash, NULL as redirect_uris FROM applications WHERE slug = ?") + sqlx::query_as::<_, Application>("SELECT id, tenant_id, name, slug, client_id, description, enabled, client_secret_hash, redirect_uris, scopes, created_at, updated_at FROM applications WHERE slug = ?") .bind(slug) .fetch_optional(&self.pool) .await } + async fn find_by_client_id(&self, client_id: &str) -> Result, sqlx::Error> { + sqlx::query_as::<_, Application>("SELECT id, tenant_id, name, slug, client_id, description, enabled, client_secret_hash, redirect_uris, scopes, created_at, updated_at FROM applications WHERE client_id = ?") + .bind(client_id) + .fetch_optional(&self.pool) + .await + } + async fn find_by_id(&self, id: &str) -> Result, sqlx::Error> { - sqlx::query_as::<_, Application>("SELECT id, tenant_id, name, slug, enabled, created_at, updated_at, NULL as description, NULL as client_secret_hash, NULL as redirect_uris FROM applications WHERE id = ?") + sqlx::query_as::<_, Application>("SELECT id, tenant_id, name, slug, client_id, description, enabled, client_secret_hash, redirect_uris, scopes, created_at, updated_at FROM applications WHERE id = ?") .bind(id) .fetch_optional(&self.pool) .await @@ -48,7 +98,7 @@ impl ApplicationsRepository for SqliteApplicationsRepository { async fn list(&self, tenant_id: &str) -> Result, sqlx::Error> { sqlx::query_as::<_, Application>( - "SELECT id, tenant_id, name, slug, enabled, created_at, updated_at, NULL as description, NULL as client_secret_hash, NULL as redirect_uris FROM applications WHERE tenant_id = ? ORDER BY name", + "SELECT id, tenant_id, name, slug, client_id, description, enabled, client_secret_hash, redirect_uris, scopes, created_at, updated_at FROM applications WHERE tenant_id = ? ORDER BY name", ) .bind(tenant_id) .fetch_all(&self.pool) @@ -57,12 +107,74 @@ impl ApplicationsRepository for SqliteApplicationsRepository { async fn set_enabled(&self, id: &str, enabled: bool) -> Result<(), sqlx::Error> { sqlx::query( - "UPDATE applications SET enabled = ?, updated_at = strftime('%Y-%m-%dT%H:%M:%SZ', 'now') WHERE id = ?", - ) - .bind(enabled) - .bind(id) - .execute(&self.pool) - .await?; + "UPDATE applications SET enabled = ?, updated_at = strftime('%Y-%m-%dT%H:%M:%SZ', 'now') WHERE id = ?", + ) + .bind(enabled) + .bind(id) + .execute(&self.pool) + .await?; + Ok(()) + } + + async fn update_secret_hash(&self, id: &str, secret_hash: &str) -> Result<(), sqlx::Error> { + sqlx::query( + "UPDATE applications SET client_secret_hash = ?, updated_at = strftime('%Y-%m-%dT%H:%M:%SZ', 'now') WHERE id = ?", + ) + .bind(secret_hash) + .bind(id) + .execute(&self.pool) + .await?; + Ok(()) + } + + async fn rotate_secret_with_audit( + &self, + id: &str, + secret_hash: &str, + audit_event: Option>, + ) -> Result<(), sqlx::Error> { + let mut tx = self.pool.begin().await?; + + let res = sqlx::query( + "UPDATE applications SET client_secret_hash = ?, updated_at = strftime('%Y-%m-%dT%H:%M:%SZ', 'now') WHERE id = ?", + ) + .bind(secret_hash) + .bind(id) + .execute(&mut *tx) + .await?; + + if res.rows_affected() == 0 { + return Err(sqlx::Error::RowNotFound); + } + + if let Some(event) = audit_event { + let audit_id = uuid::Uuid::new_v4().to_string(); + let severity_str = event.severity.to_string(); + sqlx::query( + r#" + INSERT INTO audit_logs ( + id, actor_user_id, target_user_id, + action, resource_type, resource_id, + severity, ip_address, user_agent, metadata_json + ) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + "#, + ) + .bind(&audit_id) + .bind(event.actor_id) + .bind(event.target_id) + .bind(event.action) + .bind(event.resource_type) + .bind(event.resource_id) + .bind(&severity_str) + .bind(event.ip) + .bind(event.ua) + .bind(event.metadata) + .execute(&mut *tx) + .await?; + } + + tx.commit().await?; Ok(()) } @@ -71,18 +183,24 @@ impl ApplicationsRepository for SqliteApplicationsRepository { id: &str, name: &str, slug: &str, + description: Option<&str>, + redirect_uris: Option<&str>, + scopes: Option<&str>, enabled: bool, ) -> Result<(), sqlx::Error> { sqlx::query( r#" UPDATE applications - SET name = ?, slug = ?, enabled = ?, + SET name = ?, slug = ?, description = ?, redirect_uris = ?, scopes = ?, enabled = ?, updated_at = strftime('%Y-%m-%dT%H:%M:%SZ', 'now') WHERE id = ? "#, ) .bind(name) .bind(slug) + .bind(description) + .bind(redirect_uris) + .bind(scopes) .bind(enabled) .bind(id) .execute(&self.pool) diff --git a/src/db/repository/traits.rs b/src/db/repository/traits.rs index 799213e..9ef7de7 100644 --- a/src/db/repository/traits.rs +++ b/src/db/repository/traits.rs @@ -96,22 +96,41 @@ pub trait SessionsRepository: Send + Sync { #[async_trait::async_trait] pub trait ApplicationsRepository: Send + Sync { - async fn create( + #[allow(clippy::too_many_arguments)] + async fn create_with_audit( &self, id: &str, tenant_id: &str, name: &str, slug: &str, + client_id: &str, + client_secret_hash: Option<&str>, + description: Option<&str>, + redirect_uris: Option<&str>, + scopes: Option<&str>, + audit_event: Option>, ) -> Result; async fn find_by_slug(&self, slug: &str) -> Result, sqlx::Error>; + async fn find_by_client_id(&self, client_id: &str) -> Result, sqlx::Error>; async fn find_by_id(&self, id: &str) -> Result, sqlx::Error>; async fn list(&self, tenant_id: &str) -> Result, sqlx::Error>; async fn set_enabled(&self, id: &str, enabled: bool) -> Result<(), sqlx::Error>; + async fn update_secret_hash(&self, id: &str, secret_hash: &str) -> Result<(), sqlx::Error>; + async fn rotate_secret_with_audit( + &self, + id: &str, + secret_hash: &str, + audit_event: Option>, + ) -> Result<(), sqlx::Error>; + #[allow(clippy::too_many_arguments)] async fn update( &self, id: &str, name: &str, slug: &str, + description: Option<&str>, + redirect_uris: Option<&str>, + scopes: Option<&str>, enabled: bool, ) -> Result<(), sqlx::Error>; async fn delete(&self, id: &str) -> Result<(), sqlx::Error>; diff --git a/src/identity/applications.rs b/src/identity/applications.rs index 5cbd3c8..44c6e08 100644 --- a/src/identity/applications.rs +++ b/src/identity/applications.rs @@ -1,11 +1,106 @@ +use crate::db::repository::traits::AuditRepositoryExt; use crate::{db::models::Application, error::AppError}; +use subtle::ConstantTimeEq; +pub const CLIENT_ID_PREFIX: &str = "nx9_app_"; +pub const CLIENT_SECRET_PREFIX: &str = "nx9_secret_"; + +/// Generate a new unique server-side Client ID. +pub fn generate_client_id() -> String { + let mut bytes = [0u8; 16]; + rand::RngCore::fill_bytes(&mut rand::thread_rng(), &mut bytes); + format!("{}{}", CLIENT_ID_PREFIX, hex::encode(bytes)) +} + +/// Generate a new CSPRNG Client Secret. +pub fn generate_client_secret() -> String { + let mut bytes = [0u8; 32]; + rand::RngCore::fill_bytes(&mut rand::thread_rng(), &mut bytes); + format!("{}{}", CLIENT_SECRET_PREFIX, hex::encode(bytes)) +} + +/// Hash a raw client secret string into a hex-encoded BLAKE3 digest. +pub fn hash_client_secret(raw: &str) -> String { + hex::encode(blake3::hash(raw.as_bytes()).as_bytes()) +} + +/// Hash a raw client secret string into a 32-byte BLAKE3 digest. +pub fn hash_secret_bytes(raw: &str) -> [u8; 32] { + *blake3::hash(raw.as_bytes()).as_bytes() +} + +/// Constant-time byte array comparison using subtle::ConstantTimeEq. +pub fn constant_time_compare(a: &[u8], b: &[u8]) -> bool { + if a.len() != b.len() { + return false; + } + a.ct_eq(b).into() +} + +pub fn validate_redirect_uris(uris: &[String]) -> Result<(), AppError> { + if uris.len() > 10 { + return Err(AppError::InvalidInput( + "maximum 10 redirect URIs allowed".into(), + )); + } + for uri in uris { + let trimmed = uri.trim(); + if trimmed.is_empty() { + return Err(AppError::InvalidInput( + "redirect URI cannot be empty".into(), + )); + } + if trimmed.len() > 512 { + return Err(AppError::InvalidInput( + "redirect URI exceeds maximum length of 512 characters".into(), + )); + } + let parsed = url::Url::parse(trimmed).map_err(|e| { + AppError::InvalidInput(format!("invalid redirect URI '{trimmed}': {e}")) + })?; + if parsed.fragment().is_some() { + return Err(AppError::InvalidInput(format!( + "redirect URI '{trimmed}' must not contain a fragment" + ))); + } + if !parsed.username().is_empty() || parsed.password().is_some() { + return Err(AppError::InvalidInput(format!( + "redirect URI '{trimmed}' must not contain user credentials" + ))); + } + match parsed.scheme() { + "https" => {} + "http" => { + let host = parsed.host_str().unwrap_or(""); + if host != "localhost" && host != "127.0.0.1" && host != "[::1]" && host != "::1" { + return Err(AppError::InvalidInput(format!( + "redirect URI '{trimmed}' with http scheme is only allowed for localhost/loopback development" + ))); + } + } + other => { + return Err(AppError::InvalidInput(format!( + "redirect URI '{trimmed}' has unsupported scheme '{other}'; only https (or http for localhost) is allowed" + ))); + } + } + } + Ok(()) +} + +#[allow(clippy::too_many_arguments)] pub async fn create( provider: &std::sync::Arc, tenant_id: &str, name: &str, slug: &str, -) -> Result { + description: Option<&str>, + redirect_uris: Option>, + scopes: Option>, + audit_actor_id: Option<&str>, + audit_ip: Option<&str>, + audit_ua: Option<&str>, +) -> Result<(Application, String), AppError> { let name = name.trim(); let slug = slug.trim(); if name.is_empty() || slug.is_empty() { @@ -13,6 +108,9 @@ pub async fn create( "name and slug cannot be empty".into(), )); } + if let Some(ref uris) = redirect_uris { + validate_redirect_uris(uris)?; + } if provider .applications() .find_by_slug(slug) @@ -22,12 +120,52 @@ pub async fn create( { return Err(AppError::Conflict(format!("slug '{slug}' already exists"))); } + let id = uuid::Uuid::new_v4().to_string(); - provider + let client_id = generate_client_id(); + let raw_secret = generate_client_secret(); + let secret_hash = hash_client_secret(&raw_secret); + + let redirect_json = redirect_uris.map(|v| serde_json::to_string(&v).unwrap_or_default()); + let scopes_json = scopes.map(|v| serde_json::to_string(&v).unwrap_or_default()); + + let metadata = serde_json::json!({ + "application_id": id, + "name": name, + "client_id": client_id, + }) + .to_string(); + + let audit_event = crate::audit::AuditEvent { + actor_id: audit_actor_id, + target_id: None, + action: "application.created", + resource_type: "application", + resource_id: Some(&id), + severity: crate::db::models::AuditSeverity::Info, + ip: audit_ip, + ua: audit_ua, + metadata: Some(&metadata), + }; + + let app = provider .applications() - .create(&id, tenant_id, name, slug) + .create_with_audit( + &id, + tenant_id, + name, + slug, + &client_id, + Some(&secret_hash), + description, + redirect_json.as_deref(), + scopes_json.as_deref(), + Some(audit_event), + ) .await - .map_err(AppError::Database) + .map_err(AppError::Database)?; + + Ok((app, raw_secret)) } pub async fn list( @@ -65,14 +203,108 @@ pub async fn find_by_slug( .ok_or(AppError::NotFound) } +pub async fn rotate_secret( + provider: &std::sync::Arc, + id: &str, + audit_actor_id: Option<&str>, + audit_ip: Option<&str>, + audit_ua: Option<&str>, +) -> Result { + let app = provider + .applications() + .find_by_id(id) + .await + .map_err(AppError::Database)? + .ok_or(AppError::NotFound)?; + + let raw_secret = generate_client_secret(); + let secret_hash = hash_client_secret(&raw_secret); + + let metadata = serde_json::json!({ + "application_id": id, + "name": app.name, + "client_id": app.get_client_id(), + }) + .to_string(); + + let audit_event = crate::audit::AuditEvent { + actor_id: audit_actor_id, + target_id: None, + action: "application.secret_rotated", + resource_type: "application", + resource_id: Some(id), + severity: crate::db::models::AuditSeverity::Warning, + ip: audit_ip, + ua: audit_ua, + metadata: Some(&metadata), + }; + + provider + .applications() + .rotate_secret_with_audit(id, &secret_hash, Some(audit_event)) + .await + .map_err(AppError::Database)?; + + Ok(raw_secret) +} + +pub async fn validate_client_credentials( + provider: &std::sync::Arc, + client_id: &str, + client_secret: &str, +) -> Result { + let supplied_digest = hash_secret_bytes(client_secret); + let app = provider + .applications() + .find_by_client_id(client_id) + .await + .map_err(AppError::Database)?; + + let dummy_digest = [0u8; 32]; + + let (valid_app, stored_digest_opt) = match app { + Some(ref a) if a.enabled => { + let digest_opt = a + .client_secret_hash + .as_ref() + .and_then(|h| hex::decode(h).ok()) + .and_then(|vec| <[u8; 32]>::try_from(vec).ok()); + (digest_opt.is_some(), digest_opt) + } + _ => (false, None), + }; + + let target_digest = stored_digest_opt.as_ref().unwrap_or(&dummy_digest); + let matches = constant_time_compare(&supplied_digest, target_digest); + + if valid_app && matches { + Ok(app.unwrap()) + } else { + Err(AppError::Unauthorized) + } +} + +#[allow(clippy::too_many_arguments)] pub async fn update( provider: &std::sync::Arc, id: &str, name: &str, slug: &str, + description: Option<&str>, + redirect_uris: Option>, + scopes: Option>, enabled: bool, + audit_actor_id: Option<&str>, + audit_ip: Option<&str>, + audit_ua: Option<&str>, ) -> Result { - let _ = provider.applications().find_by_id(id).await?; + let existing = provider + .applications() + .find_by_id(id) + .await + .map_err(AppError::Database)? + .ok_or(AppError::NotFound)?; + let name = name.trim(); let slug = slug.trim(); if name.is_empty() || slug.is_empty() { @@ -80,6 +312,9 @@ pub async fn update( "name and slug cannot be empty".into(), )); } + if let Some(ref uris) = redirect_uris { + validate_redirect_uris(uris)?; + } if let Some(other) = provider .applications() .find_by_slug(slug) @@ -90,40 +325,161 @@ pub async fn update( return Err(AppError::Conflict(format!("slug '{slug}' already exists"))); } } + + let redirect_json = redirect_uris.map(|v| serde_json::to_string(&v).unwrap_or_default()); + let scopes_json = scopes.map(|v| serde_json::to_string(&v).unwrap_or_default()); + provider .applications() - .update(id, name, slug, enabled) + .update( + id, + name, + slug, + description, + redirect_json.as_deref(), + scopes_json.as_deref(), + enabled, + ) .await .map_err(AppError::Database)?; - provider + + let updated = provider .applications() .find_by_id(id) .await - .map_err(crate::error::AppError::Database)? - .ok_or_else(|| crate::error::AppError::NotFound) + .map_err(AppError::Database)? + .ok_or(AppError::NotFound)?; + + let action = if existing.enabled != enabled { + if enabled { + "application.enabled" + } else { + "application.disabled" + } + } else { + "application.updated" + }; + + let metadata = serde_json::json!({ + "application_id": id, + "name": updated.name, + "client_id": updated.get_client_id(), + "enabled": enabled, + }) + .to_string(); + + provider + .audit() + .log(crate::audit::AuditEvent { + actor_id: audit_actor_id, + target_id: None, + action, + resource_type: "application", + resource_id: Some(id), + severity: crate::db::models::AuditSeverity::Info, + ip: audit_ip, + ua: audit_ua, + metadata: Some(&metadata), + }) + .await?; + + Ok(updated) } pub async fn set_enabled( provider: &std::sync::Arc, id: &str, enabled: bool, + audit_actor_id: Option<&str>, + audit_ip: Option<&str>, + audit_ua: Option<&str>, ) -> Result<(), AppError> { - let _ = provider.applications().find_by_id(id).await?; + let app = provider + .applications() + .find_by_id(id) + .await + .map_err(AppError::Database)? + .ok_or(AppError::NotFound)?; + provider .applications() .set_enabled(id, enabled) .await - .map_err(AppError::Database) + .map_err(AppError::Database)?; + + let action = if enabled { + "application.enabled" + } else { + "application.disabled" + }; + + let metadata = serde_json::json!({ + "application_id": id, + "name": app.name, + "client_id": app.get_client_id(), + "enabled": enabled, + }) + .to_string(); + + provider + .audit() + .log(crate::audit::AuditEvent { + actor_id: audit_actor_id, + target_id: None, + action, + resource_type: "application", + resource_id: Some(id), + severity: crate::db::models::AuditSeverity::Info, + ip: audit_ip, + ua: audit_ua, + metadata: Some(&metadata), + }) + .await?; + + Ok(()) } pub async fn delete( provider: &std::sync::Arc, id: &str, + audit_actor_id: Option<&str>, + audit_ip: Option<&str>, + audit_ua: Option<&str>, ) -> Result<(), AppError> { - let _ = provider.applications().find_by_id(id).await?; + let app = provider + .applications() + .find_by_id(id) + .await + .map_err(AppError::Database)? + .ok_or(AppError::NotFound)?; + provider .applications() .delete(id) .await - .map_err(AppError::Database) + .map_err(AppError::Database)?; + + let metadata = serde_json::json!({ + "application_id": id, + "name": app.name, + "client_id": app.get_client_id(), + }) + .to_string(); + + provider + .audit() + .log(crate::audit::AuditEvent { + actor_id: audit_actor_id, + target_id: None, + action: "application.deleted", + resource_type: "application", + resource_id: Some(id), + severity: crate::db::models::AuditSeverity::Warning, + ip: audit_ip, + ua: audit_ua, + metadata: Some(&metadata), + }) + .await?; + + Ok(()) } diff --git a/src/runtime/application.rs b/src/runtime/application.rs index 3f133fd..f15083d 100644 --- a/src/runtime/application.rs +++ b/src/runtime/application.rs @@ -29,6 +29,8 @@ pub struct Application { pub signals: SignalManager, pub shutdown: ShutdownCoordinator, pub metrics: RuntimeMetrics, + pub local_addr: Option, + pub bound_port: Arc, } impl Application { @@ -82,52 +84,57 @@ impl Application { &self.metrics } - /// Force a runtime state update. + /// Force advance runtime state (monotonic, forward-only). pub fn set_state(&self, state: RuntimeState) { - self.state.force_set(state); + self.state.force_advance(state); } /// Perform graceful shutdown flow explicitly. pub async fn perform_shutdown(&mut self) -> Result<()> { - if !self.state.initiate_shutdown() { - if self.state.load().is_shutting_down() { - return Ok(()); + let current_state = self.state.load(); + + if current_state == RuntimeState::Running { + let _ = self.state.initiate_shutdown(); + } else if !current_state.is_shutting_down() { + self.state.force_advance(RuntimeState::Draining); + } + + if self.state.load() == RuntimeState::Draining { + tracing::info!("draining active connections"); + let _ = self + .state + .transition(RuntimeState::Draining, RuntimeState::StoppingWorkers); + } + + if self.state.load() == RuntimeState::StoppingWorkers { + tracing::info!("stopping background workers"); + self.workers + .shutdown_all_with_coordinator(Duration::from_secs(10), Some(&self.shutdown)) + .await; + let _ = self + .state + .transition(RuntimeState::StoppingWorkers, RuntimeState::ExecutingHooks); + } + + if self.state.load() == RuntimeState::ExecutingHooks { + tracing::info!("executing shutdown hooks"); + self.hooks.execute_all().await; + let _ = self + .state + .transition(RuntimeState::ExecutingHooks, RuntimeState::ClosingResources); + } + + if self.state.load() == RuntimeState::ClosingResources { + tracing::info!("closing database connection pool and resources"); + if let Some(pool) = self.pool_handle.take() { + pool.close().await; } - self.state.force_set(RuntimeState::Draining); + let _ = self + .state + .transition(RuntimeState::ClosingResources, RuntimeState::Stopped); } - println!("Draining"); - tracing::info!("draining active connections"); - - let _ = self - .state - .transition(RuntimeState::Draining, RuntimeState::StoppingWorkers); - println!("StoppingWorkers"); - tracing::info!("stopping background workers"); - self.workers.shutdown_all(Duration::from_secs(10)).await; - - let _ = self - .state - .transition(RuntimeState::StoppingWorkers, RuntimeState::ExecutingHooks); - println!("ExecutingHooks"); - tracing::info!("executing shutdown hooks"); - self.hooks.execute_all().await; - - let _ = self - .state - .transition(RuntimeState::ExecutingHooks, RuntimeState::ClosingResources); - println!("ClosingResources"); - tracing::info!("closing database connection pool and resources"); - if let Some(pool) = self.pool_handle.take() { - pool.close().await; - } - - let _ = self - .state - .transition(RuntimeState::ClosingResources, RuntimeState::Stopped); - println!("Stopped"); tracing::info!("application stopped cleanly"); - Ok(()) } } @@ -135,12 +142,10 @@ impl Application { #[async_trait::async_trait] impl Lifecycle for Application { async fn initialize(&mut self) -> Result<()> { - println!("Initializing"); let _ = self .state .transition(RuntimeState::Initializing, RuntimeState::Starting); - println!("Starting"); let config = match &self.config { Some(cfg) => cfg.clone(), None => { @@ -179,8 +184,6 @@ impl Lifecycle for Application { .transition(RuntimeState::Starting, RuntimeState::Running); } - println!("Running"); - let config = self.config.as_ref().cloned().unwrap_or_default(); let addr_str = format!("{}:{}", config.server.host, config.server.port); let listener = tokio::net::TcpListener::bind(&addr_str) @@ -188,7 +191,9 @@ impl Lifecycle for Application { .with_context(|| format!("failed to bind TCP listener to {addr_str}"))?; let local_addr = listener.local_addr()?; - println!("Listening on {}", local_addr); + self.local_addr = Some(local_addr); + self.bound_port + .store(local_addr.port(), std::sync::atomic::Ordering::Release); tracing::info!(address = %local_addr, "Listening on {}", local_addr); let router = match self.router.take() { @@ -205,24 +210,42 @@ impl Lifecycle for Application { let signal_mgr = self.signals.clone(); let shutdown_coord = self.shutdown.clone(); + let state = self.state.clone(); - let server = axum::serve(listener, router).with_graceful_shutdown(async move { - tokio::select! { - sig = signals::wait_for_shutdown_signal() => { - tracing::info!(signal = sig, "received shutdown signal"); - signal_mgr.record_signal(); - shutdown_coord.cancel(); - } - _ = shutdown_coord.cancelled() => { - tracing::info!("shutdown coordinator cancelled"); - } - } + let signal_task = tokio::spawn(signals::listen_for_signals( + signal_mgr, + shutdown_coord.clone(), + self.state.clone(), + )); + + let graceful_token = shutdown_coord.token().clone(); + let forced_token = shutdown_coord.forced_token().clone(); + + let state_for_shutdown = state.clone(); + let server_fut = axum::serve(listener, router).with_graceful_shutdown(async move { + graceful_token.cancelled().await; + tracing::info!("graceful shutdown triggered; initiating HTTP connection draining"); + let _ = state_for_shutdown.initiate_shutdown(); }); - if let Err(err) = server.await { - tracing::error!(error = %err, "HTTP server error"); + let mut server_task = tokio::spawn(async move { server_fut.await }); + + tokio::select! { + res = &mut server_task => { + match res { + Ok(Ok(())) => tracing::info!("HTTP server stopped gracefully"), + Ok(Err(err)) => tracing::error!(error = %err, "HTTP server error"), + Err(join_err) => tracing::debug!(error = %join_err, "HTTP server task finished"), + } + } + _ = forced_token.cancelled() => { + tracing::warn!("live forced shutdown escalation received during HTTP drain; aborting server task immediately"); + server_task.abort(); + let _ = server_task.await; + } } + signal_task.abort(); self.perform_shutdown().await } diff --git a/src/runtime/cancellation.rs b/src/runtime/cancellation.rs index 5f0b0c2..ccc3aef 100644 --- a/src/runtime/cancellation.rs +++ b/src/runtime/cancellation.rs @@ -2,36 +2,76 @@ use tokio_util::sync::CancellationToken; +/// Dual-token shutdown coordinator that supports graceful termination +/// (1st signal) and live forced escalation (2nd signal). #[derive(Clone)] pub struct ShutdownCoordinator { - root: CancellationToken, + graceful: CancellationToken, + forced: CancellationToken, } impl ShutdownCoordinator { pub fn new() -> Self { Self { - root: CancellationToken::new(), + graceful: CancellationToken::new(), + forced: CancellationToken::new(), } } + /// Access the primary graceful cancellation token. pub fn token(&self) -> &CancellationToken { - &self.root + &self.graceful } + /// Access the forced cancellation token. + pub fn forced_token(&self) -> &CancellationToken { + &self.forced + } + + /// Create a child token linked to graceful cancellation. pub fn child_token(&self) -> CancellationToken { - self.root.child_token() + self.graceful.child_token() } + /// Trigger graceful shutdown. pub fn cancel(&self) { - self.root.cancel(); + self.graceful.cancel(); } + /// Trigger graceful shutdown explicitly. + pub fn cancel_graceful(&self) { + self.graceful.cancel(); + } + + /// Trigger forced shutdown escalation live. + pub fn cancel_forced(&self) { + self.graceful.cancel(); + self.forced.cancel(); + } + + /// Check if graceful shutdown has been initiated. pub fn is_cancelled(&self) -> bool { - self.root.is_cancelled() + self.graceful.is_cancelled() } + /// Check if forced escalation has been triggered. + pub fn is_forced(&self) -> bool { + self.forced.is_cancelled() + } + + /// Await graceful cancellation. pub async fn cancelled(&self) { - self.root.cancelled().await; + self.graceful.cancelled().await; + } + + /// Await graceful cancellation explicitly. + pub async fn graceful_cancelled(&self) { + self.graceful.cancelled().await; + } + + /// Await forced escalation live. + pub async fn forced_cancelled(&self) { + self.forced.cancelled().await; } } diff --git a/src/runtime/hooks.rs b/src/runtime/hooks.rs index 8ccc6de..2285534 100644 --- a/src/runtime/hooks.rs +++ b/src/runtime/hooks.rs @@ -40,7 +40,12 @@ impl HookRegistry { } let mut indices: Vec = (0..self.hooks.len()).collect(); - indices.sort_by_key(|&i| self.hooks[i].priority()); + indices.sort_by( + |&a, &b| match self.hooks[a].priority().cmp(&self.hooks[b].priority()) { + std::cmp::Ordering::Equal => a.cmp(&b), + ord => ord, + }, + ); for i in indices { let hook = &self.hooks[i]; diff --git a/src/runtime/signals.rs b/src/runtime/signals.rs index b319c80..b2d2355 100644 --- a/src/runtime/signals.rs +++ b/src/runtime/signals.rs @@ -3,6 +3,8 @@ use std::sync::Arc; use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; +use super::{AtomicRuntimeState, ShutdownCoordinator}; + #[derive(Clone)] pub struct SignalManager { signal_count: Arc, @@ -32,6 +34,17 @@ impl SignalManager { } count } + + /// Record a signal and trigger live escalation on the coordinator. + pub fn handle_signal(&self, coordinator: &ShutdownCoordinator) -> usize { + let count = self.record_signal(); + if count == 1 { + coordinator.cancel_graceful(); + } else if count >= 2 { + coordinator.cancel_forced(); + } + count + } } impl Default for SignalManager { @@ -65,3 +78,24 @@ pub async fn wait_for_shutdown_signal() -> &'static str { name = sigterm => name, } } + +/// Continuous signal monitor that remains active during graceful shutdown +/// to observe and trigger forced escalation live. +pub async fn listen_for_signals( + signal_mgr: SignalManager, + coordinator: ShutdownCoordinator, + state: AtomicRuntimeState, +) { + loop { + let sig = wait_for_shutdown_signal().await; + let count = signal_mgr.handle_signal(&coordinator); + tracing::info!(signal = sig, count, "received OS signal"); + if count == 1 { + let _ = state.initiate_shutdown(); + } else { + // 2nd signal received: forced escalation + tracing::warn!("second signal received; escalating to forced shutdown"); + break; + } + } +} diff --git a/src/runtime/state.rs b/src/runtime/state.rs index d516ba6..9d6f54a 100644 --- a/src/runtime/state.rs +++ b/src/runtime/state.rs @@ -70,19 +70,22 @@ impl fmt::Display for RuntimeState { } } +use std::sync::Arc; + /// Lock-free atomic runtime state container. /// /// Uses `AtomicU8` with `compare_exchange` to ensure deterministic, /// race-free state transitions without mutex contention. +#[derive(Clone)] pub struct AtomicRuntimeState { - state: AtomicU8, + state: Arc, } impl AtomicRuntimeState { /// Create a new state machine in the `Initializing` state. pub fn new() -> Self { Self { - state: AtomicU8::new(RuntimeState::Initializing as u8), + state: Arc::new(AtomicU8::new(RuntimeState::Initializing as u8)), } } @@ -91,15 +94,32 @@ impl AtomicRuntimeState { RuntimeState::from_u8(self.state.load(Ordering::Acquire)).unwrap_or(RuntimeState::Stopped) } + /// Check if a state transition follows the valid lifecycle graph. + pub fn is_valid_transition(expected: RuntimeState, new: RuntimeState) -> bool { + (new as u8) == (expected as u8) + 1 + } + /// Attempt an atomic state transition from `expected` to `new`. /// - /// Returns `Ok(new)` if the transition succeeded, or `Err(actual)` if the - /// current state did not match `expected`. + /// The transition is validated against the lifecycle graph. Returns `Ok(new)` + /// if the transition succeeded, or `Err(actual)` if the transition was invalid + /// or the current state did not match `expected`. pub fn transition( &self, expected: RuntimeState, new: RuntimeState, ) -> Result { + if !Self::is_valid_transition(expected, new) { + let actual = self.load(); + tracing::warn!( + expected = %expected, + actual = %actual, + target = %new, + "illegal lifecycle graph transition rejected" + ); + return Err(actual); + } + match self.state.compare_exchange( expected as u8, new as u8, @@ -123,13 +143,39 @@ impl AtomicRuntimeState { } } - /// Unconditionally advance the state. Used during forced shutdown when - /// intermediate states may have been skipped. - pub fn force_set(&self, new: RuntimeState) { - let prev = self.state.swap(new as u8, Ordering::AcqRel); - let prev_state = RuntimeState::from_u8(prev).unwrap_or(RuntimeState::Stopped); - if prev_state != new { - tracing::info!(from = %prev_state, to = %new, "runtime state forced"); + /// Unconditionally advance the state forward. Used during emergency recovery + /// when intermediate states are skipped. + /// + /// Restricted to `pub(crate)` visibility to preserve lifecycle graph invariants. + /// Guarantees monotonic forward movement (`new >= current_state`) and rejects + /// backward state regressions. + pub(crate) fn force_advance(&self, new: RuntimeState) { + loop { + let current = self.load(); + if (new as u8) < (current as u8) { + tracing::warn!( + current = %current, + target = %new, + "rejected state regression in force_advance" + ); + break; + } + if current == new { + break; + } + if self + .state + .compare_exchange( + current as u8, + new as u8, + Ordering::AcqRel, + Ordering::Acquire, + ) + .is_ok() + { + tracing::info!(from = %current, to = %new, "runtime state force advanced"); + break; + } } } @@ -180,9 +226,25 @@ mod tests { } #[test] - fn test_force_set() { + fn test_invalid_graph_transition_rejected() { let state = AtomicRuntimeState::new(); - state.force_set(RuntimeState::ClosingResources); + // Initializing -> ClosingResources is invalid in the normal graph + assert!( + state + .transition(RuntimeState::Initializing, RuntimeState::ClosingResources) + .is_err() + ); + assert_eq!(state.load(), RuntimeState::Initializing); + } + + #[test] + fn test_force_advance() { + let state = AtomicRuntimeState::new(); + state.force_advance(RuntimeState::ClosingResources); + assert_eq!(state.load(), RuntimeState::ClosingResources); + + // State regression must be rejected + state.force_advance(RuntimeState::Initializing); assert_eq!(state.load(), RuntimeState::ClosingResources); } diff --git a/src/runtime/workers.rs b/src/runtime/workers.rs index 5d1d6f1..0710891 100644 --- a/src/runtime/workers.rs +++ b/src/runtime/workers.rs @@ -6,6 +6,8 @@ use std::time::Duration; use tokio::task::JoinSet; +use super::ShutdownCoordinator; + pub struct TaskGroup { name: String, tasks: JoinSet<()>, @@ -100,11 +102,104 @@ impl WorkerManager { } } - pub async fn shutdown_all(&mut self, timeout: Duration) { + pub async fn drain_all(&mut self) { for group in self.groups.values_mut() { - group.shutdown(timeout).await; + group.abort_all(); + while group.tasks.join_next().await.is_some() {} } } + + /// Shut down all worker groups concurrently under a single global deadline, + /// while observing live forced shutdown escalation. + pub async fn shutdown_all_with_coordinator( + &mut self, + timeout: Duration, + coordinator: Option<&ShutdownCoordinator>, + ) { + let active = self.active_tasks(); + if active == 0 { + return; + } + + tracing::info!( + active_tasks = active, + timeout_secs = timeout.as_secs(), + "shutting down background worker task groups under global deadline" + ); + + let is_already_forced = coordinator.map(|c| c.is_forced()).unwrap_or(false); + if is_already_forced { + tracing::warn!("forced shutdown active; aborting all worker tasks immediately"); + self.drain_all().await; + return; + } + + let groups = std::mem::take(&mut self.groups); + let mut group_joiners = JoinSet::new(); + let mut group_map = HashMap::new(); + + for (name, mut group) in groups { + group_joiners.spawn(async move { + while group.tasks.join_next().await.is_some() {} + (name, group) + }); + } + + let join_all_fut = async { + while let Some(res) = group_joiners.join_next().await { + if let Ok((name, group)) = res { + group_map.insert(name, group); + } + } + }; + + let forced_fut = async { + if let Some(coord) = coordinator { + coord.forced_cancelled().await; + } else { + std::future::pending::<()>().await; + } + }; + + tokio::select! { + _ = join_all_fut => { + tracing::info!("all worker task groups shut down cleanly"); + } + _ = tokio::time::sleep(timeout) => { + tracing::warn!("global worker shutdown timeout expired; aborting remaining tasks"); + group_joiners.abort_all(); + while let Some(res) = group_joiners.join_next().await { + if let Ok((name, mut group)) = res { + group.abort_all(); + group_map.insert(name, group); + } + } + } + _ = forced_fut => { + tracing::warn!("live forced shutdown escalation received during worker wait; aborting remaining tasks immediately"); + group_joiners.abort_all(); + while let Some(res) = group_joiners.join_next().await { + if let Ok((name, mut group)) = res { + group.abort_all(); + group_map.insert(name, group); + } + } + } + } + + for group in group_map.values_mut() { + if !group.is_empty() { + group.abort_all(); + while group.tasks.join_next().await.is_some() {} + } + } + + self.groups = group_map; + } + + pub async fn shutdown_all(&mut self, timeout: Duration) { + self.shutdown_all_with_coordinator(timeout, None).await; + } } impl Default for WorkerManager { diff --git a/tests/application_credentials_test.rs b/tests/application_credentials_test.rs new file mode 100644 index 0000000..669ed5a --- /dev/null +++ b/tests/application_credentials_test.rs @@ -0,0 +1,722 @@ +#![cfg(feature = "sqlite")] + +use axum::{ + body::Body, + http::{Request, StatusCode, header}, +}; +use http_body_util::BodyExt; +use serde_json::Value; +use std::sync::Arc; +use tower::ServiceExt; + +use nx9_auth::{ + api, + config::{Config, DatabaseConfig, SecurityConfig, ServerConfig}, + db::{self, models::Tenant, provider::SqliteProvider}, + error::AppError, + identity::{applications, roles, users}, + security::sessions, + state::AppState, +}; + +async fn setup_test_db() -> ( + Arc, + sqlx::SqlitePool, + String, +) { + let db_id = uuid::Uuid::new_v4().to_string(); + let db_path = format!("target/test_app_{}.db", db_id); + let pool = db::create_pool(&db_path) + .await + .expect("Failed to create test pool"); + db::run_migrations(&pool) + .await + .expect("Failed to run test migrations"); + let provider = Arc::new(SqliteProvider::new(pool.clone())); + (provider, pool, db_path) +} + +async fn teardown_test_db(path: String) { + let _ = std::fs::remove_file(path); +} + +fn test_security_config() -> SecurityConfig { + SecurityConfig { + session_ttl_hours: 24, + session_absolute_ttl_days: 30, + token_ttl_days: 365, + argon2_memory: 4096, + argon2_iterations: 1, + argon2_parallelism: 1, + } +} + +fn test_config(db_path: String) -> Config { + Config { + server: ServerConfig { + host: "127.0.0.1".into(), + port: 8655, + cookie_secure: false, + production: false, + }, + database: DatabaseConfig { + path: Some(db_path), + ..Default::default() + }, + security: test_security_config(), + audit: nx9_auth::config::AuditConfig { enabled: true }, + ..Default::default() + } +} + +async fn setup_app() -> (axum::Router, String, String, String) { + let (provider, _pool, db_path) = setup_test_db().await; + let config = test_config(db_path.clone()); + let sec_cfg = config.security.clone(); + + let admin = users::create_user( + &provider, + &sec_cfg, + Tenant::DEFAULT_ID, + "admin_app_user", + "AdminSecret123!", + None, + None, + None, + ) + .await + .unwrap(); + + roles::assign_role(&provider, &admin.id, "admin", None, None, None) + .await + .unwrap(); + + let (_session, raw_token) = sessions::create_session( + &provider, + &admin.id, + Some("127.0.0.1"), + Some("TestUA"), + &sec_cfg, + ) + .await + .unwrap(); + + let state = AppState::new(provider.clone(), config); + let router = api::router::build(state); + (router, admin.id, raw_token, db_path) +} + +#[tokio::test] +async fn test_application_credential_generation_and_validation() { + let (provider, _pool, db_path) = setup_test_db().await; + + let client_id = applications::generate_client_id(); + assert!(client_id.starts_with("nx9_app_")); + assert_eq!(client_id.len(), 40); // nx9_app_ (8) + 32 hex chars = 40 + + let client_secret = applications::generate_client_secret(); + assert!(client_secret.starts_with("nx9_secret_")); + assert_eq!(client_secret.len(), 75); // nx9_secret_ (11) + 64 hex chars = 75 + + let (app, raw_secret) = applications::create( + &provider, + Tenant::DEFAULT_ID, + "Test App", + "test-app", + Some("Description of Test App"), + Some(vec!["https://example.com/callback".into()]), + Some(vec!["openid".into(), "profile".into()]), + None, + None, + None, + ) + .await + .unwrap(); + + assert!(app.get_client_id().starts_with("nx9_app_")); + assert!(app.has_credentials()); + assert_eq!(app.redirect_urls(), vec!["https://example.com/callback"]); + assert_eq!(app.scopes(), vec!["openid", "profile"]); + + // Secret hash in DB must be hex encoded BLAKE3 digest, not plaintext secret + assert_ne!(app.client_secret_hash.as_ref().unwrap(), &raw_secret); + + // Valid credentials authentication + let validated = + applications::validate_client_credentials(&provider, app.get_client_id(), &raw_secret) + .await + .unwrap(); + assert_eq!(validated.id, app.id); + + // Invalid secret + let invalid_sec = applications::validate_client_credentials( + &provider, + app.get_client_id(), + "nx9_secret_invalid", + ) + .await; + assert!(matches!(invalid_sec, Err(AppError::Unauthorized))); + + // Unknown client_id + let unknown_client = + applications::validate_client_credentials(&provider, "nx9_app_nonexistent", &raw_secret) + .await; + assert!(matches!(unknown_client, Err(AppError::Unauthorized))); + + teardown_test_db(db_path).await; +} + +#[tokio::test] +async fn test_application_secret_rotation() { + let (provider, _pool, db_path) = setup_test_db().await; + + let (app, old_secret) = applications::create( + &provider, + Tenant::DEFAULT_ID, + "Rotate App", + "rotate-app", + None, + None, + None, + None, + None, + None, + ) + .await + .unwrap(); + + let new_secret = applications::rotate_secret(&provider, &app.id, None, None, None) + .await + .unwrap(); + + assert_ne!(old_secret, new_secret); + + // Old secret fails + let old_val = + applications::validate_client_credentials(&provider, app.get_client_id(), &old_secret) + .await; + assert!(matches!(old_val, Err(AppError::Unauthorized))); + + // New secret succeeds + let new_val = + applications::validate_client_credentials(&provider, app.get_client_id(), &new_secret) + .await; + assert!(new_val.is_ok()); + + teardown_test_db(db_path).await; +} + +#[tokio::test] +async fn test_application_api_endpoints_and_cache_control() { + let (app_router, _user_id, token, db_path) = setup_app().await; + + // 1. Create Application API + let req_body = serde_json::json!({ + "name": "API Test App", + "slug": "api-test-app", + "description": "App built for API testing", + "redirect_urls": ["https://app.test/cb"], + "scopes": ["openid", "profile"] + }); + + let req = Request::builder() + .method("POST") + .uri("/api/v1/applications") + .header(header::CONTENT_TYPE, "application/json") + .header(header::COOKIE, format!("nx9_session={token}")) + .body(Body::from(serde_json::to_vec(&req_body).unwrap())) + .unwrap(); + + let resp = app_router.clone().oneshot(req).await.unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + assert_eq!( + resp.headers() + .get(header::CACHE_CONTROL) + .unwrap() + .to_str() + .unwrap(), + "no-store" + ); + + let body_bytes = resp.into_body().collect().await.unwrap().to_bytes(); + let create_resp: Value = serde_json::from_slice(&body_bytes).unwrap(); + let app_obj = &create_resp["application"]; + let client_id = app_obj["client_id"].as_str().unwrap().to_string(); + let app_id = app_obj["id"].as_str().unwrap().to_string(); + let client_secret = create_resp["client_secret"].as_str().unwrap().to_string(); + + assert!(client_id.starts_with("nx9_app_")); + assert!(client_secret.starts_with("nx9_secret_")); + + // 2. GET Application API (Must NOT expose secret or secret hash) + let get_req = Request::builder() + .method("GET") + .uri(format!("/api/v1/applications/{app_id}")) + .header(header::COOKIE, format!("nx9_session={token}")) + .body(Body::empty()) + .unwrap(); + + let get_resp = app_router.clone().oneshot(get_req).await.unwrap(); + assert_eq!(get_resp.status(), StatusCode::OK); + let get_bytes = get_resp.into_body().collect().await.unwrap().to_bytes(); + let get_json: Value = serde_json::from_slice(&get_bytes).unwrap(); + let get_app = &get_json["application"]; + + assert_eq!(get_app["client_id"], client_id); + assert!(get_app.get("client_secret").is_none()); + assert!(get_app.get("client_secret_hash").is_none()); + assert_eq!(get_app["credentials_configured"], true); + + // 3. PATCH Application containing `client_id` MUST be rejected by `deny_unknown_fields` + let patch_invalid = serde_json::json!({ + "name": "Updated Name", + "slug": "api-test-app", + "client_id": "nx9_app_hack_attempt", + "enabled": true + }); + + let patch_req = Request::builder() + .method("PATCH") + .uri(format!("/api/v1/applications/{app_id}")) + .header(header::CONTENT_TYPE, "application/json") + .header(header::COOKIE, format!("nx9_session={token}")) + .body(Body::from(serde_json::to_vec(&patch_invalid).unwrap())) + .unwrap(); + + let patch_resp = app_router.clone().oneshot(patch_req).await.unwrap(); + assert!(patch_resp.status().is_client_error()); // 400 / 422 Bad Request due to deny_unknown_fields + + // 4. Rotate Secret API + let rotate_req = Request::builder() + .method("POST") + .uri(format!("/api/v1/applications/{app_id}/secret")) + .header(header::COOKIE, format!("nx9_session={token}")) + .body(Body::empty()) + .unwrap(); + + let rotate_resp = app_router.clone().oneshot(rotate_req).await.unwrap(); + assert_eq!(rotate_resp.status(), StatusCode::OK); + assert_eq!( + rotate_resp + .headers() + .get(header::CACHE_CONTROL) + .unwrap() + .to_str() + .unwrap(), + "no-store" + ); + + let rotate_bytes = rotate_resp.into_body().collect().await.unwrap().to_bytes(); + let rotate_json: Value = serde_json::from_slice(&rotate_bytes).unwrap(); + let new_secret = rotate_json["client_secret"].as_str().unwrap(); + assert!(new_secret.starts_with("nx9_secret_")); + assert_ne!(new_secret, client_secret); + + teardown_test_db(db_path).await; +} + +#[tokio::test] +async fn test_application_database_migration_backfill_and_upgrade() { + let db_id = uuid::Uuid::new_v4().to_string(); + let db_path = format!("target/test_upgrade_{}.db", db_id); + let pool = db::create_pool(&db_path).await.unwrap(); + + // Execute migrations up to 0016 manually to simulate a v0.3.0 existing database + sqlx::query(include_str!( + "../src/db/migrations/sqlite/0001_create_tenants.sql" + )) + .execute(&pool) + .await + .unwrap(); + sqlx::query(include_str!( + "../src/db/migrations/sqlite/0002_create_users.sql" + )) + .execute(&pool) + .await + .unwrap(); + sqlx::query(include_str!( + "../src/db/migrations/sqlite/0003_create_user_profiles.sql" + )) + .execute(&pool) + .await + .unwrap(); + sqlx::query(include_str!( + "../src/db/migrations/sqlite/0004_create_roles.sql" + )) + .execute(&pool) + .await + .unwrap(); + sqlx::query(include_str!( + "../src/db/migrations/sqlite/0005_create_permissions.sql" + )) + .execute(&pool) + .await + .unwrap(); + sqlx::query(include_str!( + "../src/db/migrations/sqlite/0006_create_role_permissions.sql" + )) + .execute(&pool) + .await + .unwrap(); + sqlx::query(include_str!( + "../src/db/migrations/sqlite/0007_create_user_roles.sql" + )) + .execute(&pool) + .await + .unwrap(); + sqlx::query(include_str!( + "../src/db/migrations/sqlite/0008_create_sessions.sql" + )) + .execute(&pool) + .await + .unwrap(); + sqlx::query(include_str!( + "../src/db/migrations/sqlite/0009_create_api_tokens.sql" + )) + .execute(&pool) + .await + .unwrap(); + sqlx::query(include_str!( + "../src/db/migrations/sqlite/0010_create_service_accounts.sql" + )) + .execute(&pool) + .await + .unwrap(); + sqlx::query(include_str!( + "../src/db/migrations/sqlite/0011_create_applications.sql" + )) + .execute(&pool) + .await + .unwrap(); + sqlx::query(include_str!( + "../src/db/migrations/sqlite/0012_create_audit_logs.sql" + )) + .execute(&pool) + .await + .unwrap(); + sqlx::query(include_str!( + "../src/db/migrations/sqlite/0013_seed_default_tenant.sql" + )) + .execute(&pool) + .await + .unwrap(); + sqlx::query(include_str!( + "../src/db/migrations/sqlite/0014_seed_roles_and_permissions.sql" + )) + .execute(&pool) + .await + .unwrap(); + sqlx::query(include_str!( + "../src/db/migrations/sqlite/0015_create_refresh_tokens.sql" + )) + .execute(&pool) + .await + .unwrap(); + sqlx::query(include_str!( + "../src/db/migrations/sqlite/0016_create_groups.sql" + )) + .execute(&pool) + .await + .unwrap(); + + let legacy_id = "30000000-0000-0000-0000-000000000099"; + sqlx::query("INSERT INTO applications (id, tenant_id, name, slug) VALUES (?, '00000000-0000-0000-0000-000000000001', 'Legacy App', 'legacy-app')") + .bind(legacy_id) + .execute(&pool) + .await + .unwrap(); + + // Now run migration 0017 and 0018 + sqlx::query(include_str!( + "../src/db/migrations/sqlite/0017_update_applications_credentials.sql" + )) + .execute(&pool) + .await + .unwrap(); + sqlx::query(include_str!( + "../src/db/migrations/sqlite/0018_harden_application_credentials.sql" + )) + .execute(&pool) + .await + .unwrap(); + + let provider: Arc = Arc::new(SqliteProvider::new(pool)); + let legacy_app = applications::get(&provider, legacy_id).await.unwrap(); + + assert_eq!(legacy_app.name, "Legacy App"); + assert_eq!(legacy_app.slug.as_deref(), Some("legacy-app")); + assert!(legacy_app.get_client_id().starts_with("nx9_app_")); + assert!(!legacy_app.has_credentials()); + + // Administrator performs secret rotation to generate credentials + let generated_secret = applications::rotate_secret(&provider, &legacy_app.id, None, None, None) + .await + .unwrap(); + let updated_legacy = applications::get(&provider, legacy_id).await.unwrap(); + assert!(updated_legacy.has_credentials()); + + // Validate generated credentials + let auth_res = applications::validate_client_credentials( + &provider, + updated_legacy.get_client_id(), + &generated_secret, + ) + .await; + assert!(auth_res.is_ok()); + + teardown_test_db(db_path).await; +} + +#[tokio::test] +async fn test_application_creation_transactional_rollback_on_audit_failure() { + let (provider, _pool, db_path) = setup_test_db().await; + + // Force audit log foreign-key failure by passing invalid actor_id + let res = applications::create( + &provider, + Tenant::DEFAULT_ID, + "Rollback App", + "rollback-app", + None, + None, + None, + Some("non_existent_actor_id_fk"), + None, + None, + ) + .await; + + assert!(res.is_err()); + + // Verify application record was NOT created in DB + let app_opt = applications::find_by_slug(&provider, "rollback-app").await; + assert!(matches!(app_opt, Err(AppError::NotFound))); + + teardown_test_db(db_path).await; +} + +#[tokio::test] +async fn test_application_secret_rotation_transactional_rollback_on_audit_failure() { + let (provider, _pool, db_path) = setup_test_db().await; + + let (app, old_secret) = applications::create( + &provider, + Tenant::DEFAULT_ID, + "Rotate Rollback App", + "rotate-rollback-app", + None, + None, + None, + None, + None, + None, + ) + .await + .unwrap(); + + let orig_hash = app.client_secret_hash.clone().unwrap(); + + // Force audit insertion failure during rotation + let fail_res = applications::rotate_secret( + &provider, + &app.id, + Some("non_existent_actor_id_fk"), + None, + None, + ) + .await; + + assert!(fail_res.is_err()); + + // Assert stored client_secret_hash in DB remains UNCHANGED + let app_after_failed_rotation = applications::get(&provider, &app.id).await.unwrap(); + assert_eq!( + app_after_failed_rotation + .client_secret_hash + .as_ref() + .unwrap(), + &orig_hash + ); + + // Assert original secret STILL authenticates successfully + let orig_auth = + applications::validate_client_credentials(&provider, app.get_client_id(), &old_secret) + .await; + assert!(orig_auth.is_ok()); + + // Perform successful rotation + let new_secret = applications::rotate_secret(&provider, &app.id, None, None, None) + .await + .unwrap(); + + // Old secret fails, new secret succeeds + let old_auth = + applications::validate_client_credentials(&provider, app.get_client_id(), &old_secret) + .await; + assert!(matches!(old_auth, Err(AppError::Unauthorized))); + + let new_auth = + applications::validate_client_credentials(&provider, app.get_client_id(), &new_secret) + .await; + assert!(new_auth.is_ok()); + + teardown_test_db(db_path).await; +} + +#[tokio::test] +async fn test_application_authentication_slug_rejection() { + let (provider, _pool, db_path) = setup_test_db().await; + + let (app, secret) = applications::create( + &provider, + Tenant::DEFAULT_ID, + "Slug Reject App", + "slug-reject-app", + None, + None, + None, + None, + None, + None, + ) + .await + .unwrap(); + + // Authentication by slug MUST fail + let slug_auth = + applications::validate_client_credentials(&provider, "slug-reject-app", &secret).await; + assert!(matches!(slug_auth, Err(AppError::Unauthorized))); + + // Authentication by client_id MUST succeed + let client_id_auth = + applications::validate_client_credentials(&provider, app.get_client_id(), &secret).await; + assert!(client_id_auth.is_ok()); + + teardown_test_db(db_path).await; +} + +#[tokio::test] +async fn test_redirect_uri_structural_validation() { + let (provider, _pool, db_path) = setup_test_db().await; + + // 1. Malformed URI + let malformed = applications::create( + &provider, + Tenant::DEFAULT_ID, + "App 1", + "app-1", + None, + Some(vec!["not-a-valid-uri".into()]), + None, + None, + None, + None, + ) + .await; + assert!(matches!(malformed, Err(AppError::InvalidInput(_)))); + + // 2. Fragment URI + let fragment = applications::create( + &provider, + Tenant::DEFAULT_ID, + "App 2", + "app-2", + None, + Some(vec!["https://example.com/callback#frag".into()]), + None, + None, + None, + None, + ) + .await; + assert!(matches!(fragment, Err(AppError::InvalidInput(_)))); + + // 3. Userinfo URI + let userinfo = applications::create( + &provider, + Tenant::DEFAULT_ID, + "App 3", + "app-3", + None, + Some(vec!["https://user:pass@example.com/callback".into()]), + None, + None, + None, + None, + ) + .await; + assert!(matches!(userinfo, Err(AppError::InvalidInput(_)))); + + // 4. Non-loopback HTTP URI (must be rejected) + let non_loopback_http = applications::create( + &provider, + Tenant::DEFAULT_ID, + "App 4", + "app-4", + None, + Some(vec!["http://example.com/callback".into()]), + None, + None, + None, + None, + ) + .await; + assert!(matches!(non_loopback_http, Err(AppError::InvalidInput(_)))); + + // 5. Custom scheme (must be rejected) + let custom_scheme = applications::create( + &provider, + Tenant::DEFAULT_ID, + "App 5", + "app-5", + None, + Some(vec!["myapp://callback".into()]), + None, + None, + None, + None, + ) + .await; + assert!(matches!(custom_scheme, Err(AppError::InvalidInput(_)))); + + // 6. >10 URIs + let too_many_uris: Vec = (0..11) + .map(|i| format!("https://example{i}.com/cb")) + .collect(); + let too_many = applications::create( + &provider, + Tenant::DEFAULT_ID, + "App 6", + "app-6", + None, + Some(too_many_uris), + None, + None, + None, + None, + ) + .await; + assert!(matches!(too_many, Err(AppError::InvalidInput(_)))); + + // 7. Valid URIs (https and http loopback) + let valid = applications::create( + &provider, + Tenant::DEFAULT_ID, + "Valid App", + "valid-app", + None, + Some(vec![ + "https://app.example.com/callback".into(), + "http://127.0.0.1:8080/callback".into(), + "http://localhost:3000/callback".into(), + ]), + None, + None, + None, + None, + ) + .await; + assert!(valid.is_ok()); + + teardown_test_db(db_path).await; +} diff --git a/tests/runtime_lifecycle_test.rs b/tests/runtime_lifecycle_test.rs index 66fed4c..173d8f2 100644 --- a/tests/runtime_lifecycle_test.rs +++ b/tests/runtime_lifecycle_test.rs @@ -1,15 +1,17 @@ use std::sync::Arc; -use std::sync::atomic::{AtomicUsize, Ordering}; -use std::time::Duration; +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; +use std::time::{Duration, Instant}; use nx9_auth::config::Config; use nx9_auth::runtime::{ - Application, HookRegistry, RuntimeState, ShutdownHook, ShutdownPriority, WorkerManager, + Application, HookRegistry, Lifecycle, RuntimeState, ShutdownCoordinator, ShutdownHook, + ShutdownPriority, WorkerManager, }; struct TestHook { name: &'static str, priority: ShutdownPriority, + should_fail: bool, counter: Arc, sequence: Arc>>, } @@ -28,6 +30,9 @@ impl ShutdownHook for TestHook { self.counter.fetch_add(1, Ordering::SeqCst); let mut seq = self.sequence.lock().await; seq.push(self.name); + if self.should_fail { + anyhow::bail!("deliberate hook failure"); + } Ok(()) } } @@ -55,18 +60,21 @@ async fn test_shutdown_hook_execution_order() { let hook_last = TestHook { name: "hook_last", priority: ShutdownPriority::Last, + should_fail: false, counter: counter.clone(), sequence: sequence.clone(), }; let hook_first = TestHook { name: "hook_first", priority: ShutdownPriority::First, + should_fail: false, counter: counter.clone(), sequence: sequence.clone(), }; let hook_normal = TestHook { name: "hook_normal", priority: ShutdownPriority::Normal, + should_fail: false, counter: counter.clone(), sequence: sequence.clone(), }; @@ -85,6 +93,74 @@ async fn test_shutdown_hook_execution_order() { assert_eq!(*seq, vec!["hook_first", "hook_normal", "hook_last"]); } +#[tokio::test] +async fn test_same_priority_hook_registration_order() { + let counter = Arc::new(AtomicUsize::new(0)); + let sequence = Arc::new(tokio::sync::Mutex::new(Vec::new())); + + let hook_n1 = TestHook { + name: "normal_1", + priority: ShutdownPriority::Normal, + should_fail: false, + counter: counter.clone(), + sequence: sequence.clone(), + }; + let hook_n2 = TestHook { + name: "normal_2", + priority: ShutdownPriority::Normal, + should_fail: false, + counter: counter.clone(), + sequence: sequence.clone(), + }; + let hook_n3 = TestHook { + name: "normal_3", + priority: ShutdownPriority::Normal, + should_fail: false, + counter: counter.clone(), + sequence: sequence.clone(), + }; + + let mut registry = HookRegistry::new(); + registry.register(Box::new(hook_n1)); + registry.register(Box::new(hook_n2)); + registry.register(Box::new(hook_n3)); + + registry.execute_all().await; + let seq = sequence.lock().await; + assert_eq!(*seq, vec!["normal_1", "normal_2", "normal_3"]); +} + +#[tokio::test] +async fn test_hook_failure_resilience() { + let counter = Arc::new(AtomicUsize::new(0)); + let sequence = Arc::new(tokio::sync::Mutex::new(Vec::new())); + + let failing_hook = TestHook { + name: "failing_hook", + priority: ShutdownPriority::Normal, + should_fail: true, + counter: counter.clone(), + sequence: sequence.clone(), + }; + let succeeding_hook = TestHook { + name: "succeeding_hook", + priority: ShutdownPriority::Normal, + should_fail: false, + counter: counter.clone(), + sequence: sequence.clone(), + }; + + let mut registry = HookRegistry::new(); + registry.register(Box::new(failing_hook)); + registry.register(Box::new(succeeding_hook)); + + registry.execute_all().await; + + assert_eq!(counter.load(Ordering::SeqCst), 2); + let seq = sequence.lock().await; + assert_eq!(*seq, vec!["failing_hook", "succeeding_hook"]); +} + #[tokio::test] async fn test_worker_manager_lifecycle() { let mut mgr = WorkerManager::new(); @@ -102,3 +178,154 @@ async fn test_worker_manager_lifecycle() { assert_eq!(mgr.active_tasks(), 0); assert_eq!(counter.load(Ordering::SeqCst), 1); } + +#[tokio::test] +async fn test_worker_live_forced_escalation_abort() { + let mut mgr = WorkerManager::new(); + let group = mgr.group("long-worker"); + + let worker_started = Arc::new(AtomicBool::new(false)); + let started = worker_started.clone(); + + group.spawn(async move { + started.store(true, Ordering::SeqCst); + tokio::time::sleep(Duration::from_secs(10)).await; + }); + + // Wait for worker to begin execution + while !worker_started.load(Ordering::SeqCst) { + tokio::time::sleep(Duration::from_millis(5)).await; + } + + assert_eq!(mgr.active_tasks(), 1); + + let coord = ShutdownCoordinator::new(); + let coord_clone = coord.clone(); + + let start_time = Instant::now(); + + let shutdown_handle = tokio::spawn(async move { + let mut m = mgr; + m.shutdown_all_with_coordinator(Duration::from_secs(10), Some(&coord_clone)) + .await; + m + }); + + // Short delay to ensure shutdown_all is actively waiting + tokio::time::sleep(Duration::from_millis(30)).await; + + // Trigger live second-signal forced escalation + coord.cancel_forced(); + + let mgr_after = shutdown_handle.await.expect("shutdown task join"); + let elapsed = start_time.elapsed(); + + assert_eq!(mgr_after.active_tasks(), 0); + assert!( + elapsed < Duration::from_millis(1000), + "Forced shutdown took {:?}, expected < 1s", + elapsed + ); +} + +#[tokio::test] +async fn test_worker_global_deadline_budget_across_groups() { + let mut mgr = WorkerManager::new(); + mgr.group("group-a").spawn(async { + tokio::time::sleep(Duration::from_secs(10)).await; + }); + mgr.group("group-b").spawn(async { + tokio::time::sleep(Duration::from_secs(10)).await; + }); + mgr.group("group-c").spawn(async { + tokio::time::sleep(Duration::from_secs(10)).await; + }); + + assert_eq!(mgr.active_tasks(), 3); + + let start = Instant::now(); + mgr.shutdown_all(Duration::from_millis(200)).await; + let elapsed = start.elapsed(); + + assert_eq!(mgr.active_tasks(), 0); + assert!( + elapsed < Duration::from_millis(800), + "Worker budget timeout across 3 groups took {:?}, expected single global deadline (~200ms)", + elapsed + ); +} + +#[tokio::test] +async fn test_forced_http_draining_escalation() -> anyhow::Result<()> { + use axum::routing::get; + + let router = axum::Router::new().route( + "/slow", + get(|| async { + tokio::time::sleep(Duration::from_secs(10)).await; + "done" + }), + ); + + let mut config = Config::default(); + config.server.host = "127.0.0.1".to_string(); + config.server.port = 0; + config.database.url = Some("sqlite::memory:".to_string()); + + let mut app = Application::builder(config).build().await?; + app.router = Some(router); + + let coord = app.shutdown_coordinator().clone(); + let state_ref = app.state.clone(); + let port_ref = app.bound_port.clone(); + + let app_task = tokio::spawn(async move { app.start().await }); + + // Wait for server task to bind and store bound_port + while port_ref.load(Ordering::Acquire) == 0 { + tokio::time::sleep(Duration::from_millis(5)).await; + } + let port = port_ref.load(Ordering::Acquire); + + // Send HTTP request to /slow in background task (will take 10s if not aborted) + let req_task = tokio::spawn(async move { + if let Ok(mut stream) = tokio::net::TcpStream::connect(format!("127.0.0.1:{port}")).await { + use tokio::io::AsyncWriteExt; + let _ = stream + .write_all(b"GET /slow HTTP/1.1\r\nHost: 127.0.0.1\r\n\r\n") + .await; + use tokio::io::AsyncReadExt; + let mut buf = [0u8; 1024]; + let _ = stream.read(&mut buf).await; + } + }); + + // Short delay for request to arrive at server + tokio::time::sleep(Duration::from_millis(50)).await; + + // Trigger 1st signal (graceful shutdown) + coord.cancel_graceful(); + + // Allow Tokio task executor to process cancellation and transition to Draining + tokio::time::sleep(Duration::from_millis(10)).await; + + // Verify RuntimeState is Draining while request is in-flight + assert_eq!(state_ref.load(), RuntimeState::Draining); + + // Trigger 2nd signal (forced escalation) + let start = Instant::now(); + coord.cancel_forced(); + + let res = app_task.await?; + let elapsed = start.elapsed(); + + assert!(res.is_ok()); + assert!( + elapsed < Duration::from_millis(1000), + "Forced HTTP shutdown took {:?}, expected < 1s", + elapsed + ); + + req_task.abort(); + Ok(()) +} diff --git a/ui/src/models/mod.rs b/ui/src/models/mod.rs index ccd819e..1fe791e 100644 --- a/ui/src/models/mod.rs +++ b/ui/src/models/mod.rs @@ -124,8 +124,12 @@ pub struct ApplicationView { #[serde(default)] pub client_id: String, #[serde(default)] + pub description: Option, + #[serde(default)] pub enabled: bool, #[serde(default)] + pub credentials_configured: bool, + #[serde(default)] pub redirect_urls: Vec, #[serde(default)] pub scopes: Vec, @@ -141,6 +145,17 @@ pub struct ApplicationsResponse { pub applications: Vec, } +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, Default)] +pub struct CreateApplicationResponse { + pub application: ApplicationView, + pub client_secret: String, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, Default)] +pub struct RotateSecretResponse { + pub client_secret: String, +} + #[derive(Debug, Clone, PartialEq, Serialize, Deserialize, Default)] pub struct ServiceAccountView { pub id: String, diff --git a/ui/src/pages/applications/mod.rs b/ui/src/pages/applications/mod.rs index 7bbd5c4..c93029d 100644 --- a/ui/src/pages/applications/mod.rs +++ b/ui/src/pages/applications/mod.rs @@ -16,7 +16,7 @@ use dioxus::prelude::*; pub fn ApplicationsPage() -> Element { let state = use_context::(); let auth = state.auth; - let can_manage = auth().has_permission("roles:manage"); + let can_manage = auth().has_permission("applications:manage"); let mut apps = use_signal(Vec::::new); let mut error = use_signal(|| Option::::None); @@ -29,6 +29,12 @@ pub fn ApplicationsPage() -> Element { let mut show_create = use_signal(|| false); let mut name = use_signal(String::new); let mut slug = use_signal(String::new); + let mut description = use_signal(String::new); + let mut redirect_urls_raw = use_signal(String::new); + let mut scopes_raw = use_signal(String::new); + + let mut one_time_secret = use_signal(|| Option::<(ApplicationView, String)>::None); + let mut rotate_target = use_signal(|| Option::::None); let mut delete_target = use_signal(|| Option::::None); let reload = use_callback(move |_: ()| { @@ -50,7 +56,7 @@ pub fn ApplicationsPage() -> Element { let mut filtered: Vec<_> = apps() .into_iter() - .filter(|a| matches_query(&a.name, &query()) || matches_query(&a.slug, &query())) + .filter(|a| matches_query(&a.name, &query()) || matches_query(&a.slug, &query()) || matches_query(&a.client_id, &query())) .collect(); let sk = sort_key(); filtered.sort_by(|a, b| match sk.as_str() { @@ -123,10 +129,16 @@ pub fn ApplicationsPage() -> Element { for a in page_items { { let app = a.clone(); - let app2 = a.clone(); + let app_rotate = a.clone(); + let app_delete = a.clone(); rsx! { tr { key: "{a.id}", - td { strong { "{a.name}" } } + td { + strong { "{a.name}" } + if let Some(desc) = &a.description { + div { class: "text-muted", style: "font-size: 0.8rem;", "{desc}" } + } + } td { code { "{a.client_id}" } } td { class: "text-muted", if a.redirect_urls.is_empty() { "—" } else { "{a.redirect_urls.join(\", \")}" } @@ -142,17 +154,26 @@ pub fn ApplicationsPage() -> Element { td { "{format_datetime(&a.created_at)}" } if can_manage { td { style: "text-align: right;", - div { class: "actions", + div { class: "actions", style: "display: inline-flex; gap: 0.25rem;", + button { + class: "btn btn-sm btn-outline", + r#type: "button", + onclick: move |_| rotate_target.set(Some(app_rotate.clone())), + "Rotate Secret" + } button { class: "btn btn-sm btn-outline", r#type: "button", onclick: move |_| { let id = app.id.clone(); - let name = app.name.clone(); - let slug = app.slug.clone(); + let n = app.name.clone(); + let s = app.slug.clone(); + let desc = app.description.clone(); + let r_urls = if app.redirect_urls.is_empty() { None } else { Some(app.redirect_urls.clone()) }; + let sc = if app.scopes.is_empty() { None } else { Some(app.scopes.clone()) }; let enabled = !app.enabled; spawn(async move { - match api::update_application(&id, &name, &slug, enabled).await { + match api::update_application(&id, &n, &s, desc.as_deref(), r_urls, sc, enabled).await { Ok(_) => { state.toast(ToastKind::Success, if enabled { "Enabled" } else { "Disabled" }); reload.call(()); @@ -166,7 +187,7 @@ pub fn ApplicationsPage() -> Element { button { class: "btn btn-sm btn-danger", r#type: "button", - onclick: move |_| delete_target.set(Some(app2.clone())), + onclick: move |_| delete_target.set(Some(app_delete.clone())), "Delete" } } @@ -189,16 +210,30 @@ pub fn ApplicationsPage() -> Element { oninput: move |v: String| { name.set(v.clone()); if slug().is_empty() || slug() == slugify(&name()) { - // keep in sync when empty-ish + slug.set(slugify(&v)); } - slug.set(slugify(&v)); }, } TextInput { - label: "Slug / Client ID", + label: "Slug", value: slug(), oninput: move |v| slug.set(v), } + TextInput { + label: "Description (optional)", + value: description(), + oninput: move |v| description.set(v), + } + TextInput { + label: "Redirect URLs (comma separated, optional)", + value: redirect_urls_raw(), + oninput: move |v| redirect_urls_raw.set(v), + } + TextInput { + label: "Allowed Scopes (comma separated, optional)", + value: scopes_raw(), + oninput: move |v| scopes_raw.set(v), + } div { class: "modal-footer", style: "margin-top:1rem; padding:0; border:none; background:transparent;", button { class: "btn btn-outline", r#type: "button", onclick: move |_| show_create.set(false), "Cancel" } @@ -207,13 +242,28 @@ pub fn ApplicationsPage() -> Element { onclick: move |_| { let n = name(); let s = slug(); + let d = if description().trim().is_empty() { None } else { Some(description().trim().to_string()) }; + let r_urls = if redirect_urls_raw().trim().is_empty() { + None + } else { + Some(redirect_urls_raw().split(',').map(|x| x.trim().to_string()).filter(|x| !x.is_empty()).collect::>()) + }; + let sc = if scopes_raw().trim().is_empty() { + None + } else { + Some(scopes_raw().split(',').map(|x| x.trim().to_string()).filter(|x| !x.is_empty()).collect::>()) + }; spawn(async move { - match api::create_application(&n, &s).await { - Ok(_) => { - state.toast(ToastKind::Success, "Application created"); + match api::create_application(&n, &s, d.as_deref(), r_urls, sc).await { + Ok(res) => { + state.toast(ToastKind::Success, "Application registered successfully"); show_create.set(false); name.set(String::new()); slug.set(String::new()); + description.set(String::new()); + redirect_urls_raw.set(String::new()); + scopes_raw.set(String::new()); + one_time_secret.set(Some((res.application, res.client_secret))); reload.call(()); } Err(e) => state.toast(ToastKind::Error, e.to_string()), @@ -225,6 +275,71 @@ pub fn ApplicationsPage() -> Element { } } + if let Some((app, sec)) = one_time_secret() { + Modal { + title: "Client Credentials Disclosed".to_string(), + open: true, + on_close: move |_| one_time_secret.set(None), + div { class: "alert alert-warning", style: "margin-bottom: 1rem; padding: 0.75rem; border-radius: 4px; background: #fff3cd; color: #856404; border: 1px solid #ffeeba;", + strong { "Important: " } + "Store this client secret securely. It will never be displayed again after closing this dialog." + } + div { style: "display: flex; flex-direction: column; gap: 0.75rem;", + div { + label { style: "font-weight: 600; display: block; font-size: 0.85rem;", "Application Name" } + div { "{app.name}" } + } + div { + label { style: "font-weight: 600; display: block; font-size: 0.85rem;", "Client ID" } + div { style: "display: flex; gap: 0.5rem; align-items: center;", + code { style: "flex: 1; padding: 0.4rem; background: #f8f9fa; border: 1px solid #e9ecef; border-radius: 4px;", "{app.client_id}" } + } + } + div { + label { style: "font-weight: 600; display: block; font-size: 0.85rem;", "Client Secret" } + div { style: "display: flex; gap: 0.5rem; align-items: center;", + code { style: "flex: 1; padding: 0.4rem; background: #f8f9fa; border: 1px solid #e9ecef; border-radius: 4px; color: #d63384; word-break: break-all;", "{sec}" } + } + } + } + div { class: "modal-footer", style: "margin-top:1.5rem; padding:0; border:none; background:transparent; justify-content: flex-end;", + button { + class: "btn btn-primary", r#type: "button", + onclick: move |_| one_time_secret.set(None), + "I have saved my secret" + } + } + } + } + + ConfirmDialog { + title: "Rotate Client Secret".to_string(), + message: format!( + "Are you sure you want to rotate the client secret for \"{}\"? Any existing client using the current secret will be invalidated immediately.", + rotate_target().as_ref().map(|a| a.name.as_str()).unwrap_or("") + ), + open: rotate_target().is_some(), + confirm_label: "Rotate Secret", + danger: true, + on_confirm: move |_| { + if let Some(a) = rotate_target() { + let target_app = a.clone(); + spawn(async move { + match api::rotate_application_secret(&target_app.id).await { + Ok(new_sec) => { + state.toast(ToastKind::Success, "Client secret rotated"); + rotate_target.set(None); + one_time_secret.set(Some((target_app, new_sec))); + reload.call(()); + } + Err(e) => state.toast(ToastKind::Error, e.to_string()), + } + }); + } + }, + on_cancel: move |_| rotate_target.set(None), + } + ConfirmDialog { title: "Delete application".to_string(), message: format!( diff --git a/ui/src/services/api.rs b/ui/src/services/api.rs index 2d36ba1..2496750 100644 --- a/ui/src/services/api.rs +++ b/ui/src/services/api.rs @@ -59,9 +59,25 @@ fn client() -> Client { Client::new() } +pub trait RequestBuilderExtHelper { + fn with_credentials_include(self) -> Self; +} + +impl RequestBuilderExtHelper for reqwest::RequestBuilder { + #[cfg(target_arch = "wasm32")] + fn with_credentials_include(self) -> Self { + self.fetch_credentials_include() + } + + #[cfg(not(target_arch = "wasm32"))] + fn with_credentials_include(self) -> Self { + self + } +} + /// Attach credentials + optional bearer session token. fn authorize(builder: reqwest::RequestBuilder) -> reqwest::RequestBuilder { - let builder = builder.fetch_credentials_include(); + let builder = builder.with_credentials_include(); if let Some(token) = session::load_access_token() { builder.header("Authorization", format!("Bearer {token}")) } else { @@ -180,7 +196,7 @@ pub async fn login(username: &str, password: &str) -> Result Result, ApiError> { Ok(r.applications) } -pub async fn create_application(name: &str, slug: &str) -> Result { - let body = serde_json::json!({ "name": name, "slug": slug }); - let r: Value = post_json("/applications", &body).await?; - serde_json::from_value(r.get("application").cloned().unwrap_or(Value::Null)) - .map_err(|e| ApiError::Other(e.to_string())) +pub async fn create_application( + name: &str, + slug: &str, + description: Option<&str>, + redirect_urls: Option>, + scopes: Option>, +) -> Result { + let body = serde_json::json!({ + "name": name, + "slug": slug, + "description": description, + "redirect_urls": redirect_urls, + "scopes": scopes, + }); + post_json("/applications", &body).await } pub async fn update_application( id: &str, name: &str, slug: &str, + description: Option<&str>, + redirect_urls: Option>, + scopes: Option>, enabled: bool, ) -> Result { - let body = serde_json::json!({ "name": name, "slug": slug, "enabled": enabled }); + let body = serde_json::json!({ + "name": name, + "slug": slug, + "description": description, + "redirect_urls": redirect_urls, + "scopes": scopes, + "enabled": enabled, + }); let r: Value = patch_json(&format!("/applications/{id}"), &body).await?; serde_json::from_value(r.get("application").cloned().unwrap_or(Value::Null)) .map_err(|e| ApiError::Other(e.to_string())) } +pub async fn rotate_application_secret(id: &str) -> Result { + let r: RotateSecretResponse = + post_json(&format!("/applications/{id}/secret"), &serde_json::json!({})).await?; + Ok(r.client_secret) +} + pub async fn delete_application(id: &str) -> Result<(), ApiError> { let _: Value = delete_json(&format!("/applications/{id}")).await?; Ok(())