Compare commits
1 Commits
feat/B5-to
...
feat/B3-co
| Author | SHA1 | Date | |
|---|---|---|---|
a9d7382be8
|
125
Cargo.lock
generated
125
Cargo.lock
generated
@@ -153,18 +153,6 @@ version = "1.0.102"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "7f202df86484c868dbad7eaa557ef785d5c66295e41b460ef922eca0723b842c"
|
||||
|
||||
[[package]]
|
||||
name = "argon2"
|
||||
version = "0.5.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "3c3610892ee6e0cbce8ae2700349fcf8f98adb0dbfbee85aec3c9179d29cc072"
|
||||
dependencies = [
|
||||
"base64ct",
|
||||
"blake2",
|
||||
"cpufeatures",
|
||||
"password-hash",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "async-channel"
|
||||
version = "2.5.0"
|
||||
@@ -439,15 +427,6 @@ dependencies = [
|
||||
"serde_core",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "blake2"
|
||||
version = "0.10.6"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "46502ad458c9a52b69d4d4d32775c788b7a1b85e8bc9d482d92250fc0e3f8efe"
|
||||
dependencies = [
|
||||
"digest",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "block-buffer"
|
||||
version = "0.10.4"
|
||||
@@ -1223,22 +1202,6 @@ dependencies = [
|
||||
"serde",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "email-encoding"
|
||||
version = "0.4.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "9298e6504d9b9e780ed3f7dfd43a61be8cd0e09eb07f7706a945b0072b6670b6"
|
||||
dependencies = [
|
||||
"base64 0.22.1",
|
||||
"memchr",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "email_address"
|
||||
version = "0.2.9"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "e079f19b08ca6239f47f8ba8509c11cf3ea30095831f7fed61441475edd8c449"
|
||||
|
||||
[[package]]
|
||||
name = "encode_unicode"
|
||||
version = "1.0.0"
|
||||
@@ -1333,7 +1296,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "74fef4569247a5f429d9156b9d0a2599914385dd189c539334c625d8099d90ab"
|
||||
dependencies = [
|
||||
"futures-core",
|
||||
"nom 7.1.3",
|
||||
"nom",
|
||||
"pin-project-lite",
|
||||
]
|
||||
|
||||
@@ -2116,15 +2079,11 @@ name = "helexa-upstream"
|
||||
version = "0.1.16"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"argon2",
|
||||
"axum",
|
||||
"chrono",
|
||||
"clap",
|
||||
"cortex-core",
|
||||
"figment",
|
||||
"jsonwebtoken",
|
||||
"lettre",
|
||||
"rand 0.8.6",
|
||||
"reqwest",
|
||||
"serde",
|
||||
"serde_json",
|
||||
@@ -2605,21 +2564,6 @@ dependencies = [
|
||||
"serde_json",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "jsonwebtoken"
|
||||
version = "9.3.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "5a87cc7a48537badeae96744432de36f4be2b4a34a05a5ef32e9dd8a1c169dde"
|
||||
dependencies = [
|
||||
"base64 0.22.1",
|
||||
"js-sys",
|
||||
"pem",
|
||||
"ring",
|
||||
"serde",
|
||||
"serde_json",
|
||||
"simple_asn1",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "lazy_static"
|
||||
version = "1.5.0"
|
||||
@@ -2635,33 +2579,6 @@ version = "0.1.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "09edd9e8b54e49e587e4f6295a7d29c3ea94d469cb40ab8ca70b288248a81db2"
|
||||
|
||||
[[package]]
|
||||
name = "lettre"
|
||||
version = "0.11.22"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "0da65617f6cb926332d039cb578aad56178da86e128db6a1b09f4c94fa5b3349"
|
||||
dependencies = [
|
||||
"async-trait",
|
||||
"base64 0.22.1",
|
||||
"email-encoding",
|
||||
"email_address",
|
||||
"fastrand",
|
||||
"futures-io",
|
||||
"futures-util",
|
||||
"httpdate",
|
||||
"idna",
|
||||
"mime",
|
||||
"nom 8.0.0",
|
||||
"percent-encoding",
|
||||
"quoted_printable",
|
||||
"rustls",
|
||||
"socket2",
|
||||
"tokio",
|
||||
"tokio-rustls",
|
||||
"url",
|
||||
"webpki-roots 1.0.7",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "libc"
|
||||
version = "0.2.185"
|
||||
@@ -3010,15 +2927,6 @@ dependencies = [
|
||||
"minimal-lexical",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "nom"
|
||||
version = "8.0.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "df9761775871bdef83bee530e60050f7e54b1105350d6884eb0fb4f46c2f9405"
|
||||
dependencies = [
|
||||
"memchr",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "nu-ansi-term"
|
||||
version = "0.50.3"
|
||||
@@ -3254,17 +3162,6 @@ dependencies = [
|
||||
"windows-link",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "password-hash"
|
||||
version = "0.5.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "346f04948ba92c43e8469c1ee6736c7563d71012b17d40745260fe106aac2166"
|
||||
dependencies = [
|
||||
"base64ct",
|
||||
"rand_core 0.6.4",
|
||||
"subtle",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "paste"
|
||||
version = "1.0.15"
|
||||
@@ -3612,12 +3509,6 @@ dependencies = [
|
||||
"proc-macro2",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "quoted_printable"
|
||||
version = "0.5.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "478e0585659a122aa407eb7e3c0e1fa51b1d8a870038bd29f0cf4a8551eea972"
|
||||
|
||||
[[package]]
|
||||
name = "r-efi"
|
||||
version = "5.3.0"
|
||||
@@ -4379,18 +4270,6 @@ version = "0.3.9"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "703d5c7ef118737c72f1af64ad2f6f8c5e1921f818cdcb97b8fe6fc69bf66214"
|
||||
|
||||
[[package]]
|
||||
name = "simple_asn1"
|
||||
version = "0.6.4"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "0d585997b0ac10be3c5ee635f1bab02d512760d14b7c468801ac8a01d9ae5f1d"
|
||||
dependencies = [
|
||||
"num-bigint",
|
||||
"num-traits",
|
||||
"thiserror 2.0.18",
|
||||
"time",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "sketches-ddsketch"
|
||||
version = "0.3.1"
|
||||
@@ -4459,7 +4338,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "5851699c4033c63636f7ea4cf7b7c1f1bf06d0cc03cfb42e711de5a5c46cf326"
|
||||
dependencies = [
|
||||
"base64 0.13.1",
|
||||
"nom 7.1.3",
|
||||
"nom",
|
||||
"serde",
|
||||
"unicode-segmentation",
|
||||
]
|
||||
|
||||
@@ -90,3 +90,18 @@ account_id = "operator"
|
||||
key_id = "infra"
|
||||
# No hard_cap → uncapped operator infra key (own fleet, own use). Still
|
||||
# metered for visibility.
|
||||
|
||||
# -- Upstream (helexa mesh) entitlements client (#57) --------------------
|
||||
# When enabled, a bearer key NOT found in [[entitlements.keys]] above is
|
||||
# validated against the helexa-upstream authority (mesh accounts), and its
|
||||
# budget is reserved/settled there. Operator-local keys (incl. the infra
|
||||
# key) never leave this process. Fail-closed: if upstream is unreachable a
|
||||
# request is refused (503 + Retry-After), never served un-authorized.
|
||||
# Disabled by default — a standalone operator runs purely local.
|
||||
[upstream]
|
||||
enabled = false
|
||||
# url = "https://upstream.helexa.ai"
|
||||
# Shared client bearer this cortex presents (maps to an operator_id
|
||||
# upstream). Override via CORTEX_UPSTREAM__BEARER in prod.
|
||||
# bearer = "replace-with-operator-client-secret"
|
||||
# timeout_secs = 5
|
||||
|
||||
@@ -22,6 +22,36 @@ pub struct GatewayConfig {
|
||||
/// setups keep working until keys are configured.
|
||||
#[serde(default)]
|
||||
pub entitlements: EntitlementsConfig,
|
||||
/// helexa-upstream client (#57). When enabled, keys not found in the
|
||||
/// local `[entitlements]` config are validated against the mesh
|
||||
/// authority, and budget is reserved/settled there. Disabled by default
|
||||
/// — a single operator runs purely local.
|
||||
#[serde(default)]
|
||||
pub upstream: UpstreamClientConfig,
|
||||
}
|
||||
|
||||
/// `[upstream]` — the helexa-upstream authority client (#57). Locally
|
||||
/// unrecognised bearer keys are resolved against `url`'s `/authz/v1` surface
|
||||
/// (mesh accounts); local keys (operator + infra) never leave the process.
|
||||
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
|
||||
pub struct UpstreamClientConfig {
|
||||
/// Enable the upstream fallthrough. Off → purely local entitlements.
|
||||
#[serde(default)]
|
||||
pub enabled: bool,
|
||||
/// Base URL of helexa-upstream (e.g. "https://upstream.helexa.ai").
|
||||
#[serde(default)]
|
||||
pub url: String,
|
||||
/// Shared client bearer this cortex presents to `/authz/v1` (maps to an
|
||||
/// operator_id upstream). Sent as `Authorization: Bearer <bearer>`.
|
||||
#[serde(default)]
|
||||
pub bearer: String,
|
||||
/// Per-call timeout (seconds) to upstream.
|
||||
#[serde(default = "default_upstream_timeout")]
|
||||
pub timeout_secs: u64,
|
||||
}
|
||||
|
||||
fn default_upstream_timeout() -> u64 {
|
||||
5
|
||||
}
|
||||
|
||||
/// `[entitlements]` — the local/static [`crate::entitlements::EntitlementProvider`]
|
||||
@@ -129,6 +159,7 @@ impl Default for GatewayConfig {
|
||||
neurons: vec![],
|
||||
models_config: default_models_path(),
|
||||
entitlements: EntitlementsConfig::default(),
|
||||
upstream: UpstreamClientConfig::default(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -81,12 +81,19 @@ pub struct BudgetSnapshot {
|
||||
pub reserved: u64,
|
||||
}
|
||||
|
||||
/// Authentication failure — the bearer key could not be resolved. Maps to
|
||||
/// `401 invalid_api_key` (#49/#63).
|
||||
/// Authentication failure — the bearer key could not be resolved.
|
||||
#[derive(Debug, thiserror::Error)]
|
||||
pub enum AuthError {
|
||||
/// The key is genuinely unknown → `401 invalid_api_key` (#49/#63).
|
||||
#[error("invalid or unknown API key")]
|
||||
InvalidKey,
|
||||
/// The authority that could resolve the key is unreachable (e.g. the
|
||||
/// helexa-upstream client failed, #57). Fail **closed** but distinctly:
|
||||
/// a transient outage must surface as `503 service_unavailable` +
|
||||
/// `Retry-After`, never `401` — a real key must not be rejected as
|
||||
/// invalid during an upstream blip.
|
||||
#[error("entitlement authority unavailable; retry in {retry_after_secs}s")]
|
||||
Unavailable { retry_after_secs: u64 },
|
||||
}
|
||||
|
||||
/// Why a reservation was refused. Carries enough for the caller to build the
|
||||
|
||||
@@ -22,7 +22,7 @@ use axum::http::header::AUTHORIZATION;
|
||||
use axum::http::{HeaderMap, HeaderValue};
|
||||
use axum::middleware::Next;
|
||||
use axum::response::Response;
|
||||
use cortex_core::entitlements::{HEADER_ACCOUNT_ID, HEADER_KEY_ID};
|
||||
use cortex_core::entitlements::{AuthError, HEADER_ACCOUNT_ID, HEADER_KEY_ID};
|
||||
use cortex_core::error_envelope::OpenAiError;
|
||||
use std::sync::Arc;
|
||||
|
||||
@@ -83,14 +83,25 @@ pub async fn require_principal(
|
||||
req.extensions_mut().insert(principal);
|
||||
next.run(req).await
|
||||
}
|
||||
// An unrecognized key only hard-fails when auth is *required*.
|
||||
// In allow-anonymous mode (the default) we must IGNORE it and
|
||||
// serve the request unauthenticated — otherwise the placeholder
|
||||
// keys that OpenAI-compatible clients send by default (opencode,
|
||||
// Open WebUI, Agent Zero, litellm) would all break, even though
|
||||
// the operator never opted into auth. Pre-#49 the bearer was
|
||||
// never inspected at all; this preserves that for require_auth=false.
|
||||
Err(_) => {
|
||||
// The entitlement authority is unreachable (upstream client
|
||||
// blip, #57). Fail **closed but distinct**: a transient outage
|
||||
// must not reject a real key as `401 invalid_api_key` — it's a
|
||||
// retryable `503`. This holds regardless of require_auth: we
|
||||
// can't safely serve a key we couldn't authorize.
|
||||
Err(AuthError::Unavailable { retry_after_secs }) => {
|
||||
envelope_response(OpenAiError::service_unavailable(
|
||||
"entitlement authority temporarily unavailable",
|
||||
Some(retry_after_secs),
|
||||
))
|
||||
}
|
||||
// A genuinely unrecognized key only hard-fails when auth is
|
||||
// *required*. In allow-anonymous mode (the default) we IGNORE it
|
||||
// and serve unauthenticated — otherwise the placeholder keys that
|
||||
// OpenAI-compatible clients send by default (opencode, Open WebUI,
|
||||
// Agent Zero, litellm) would all break though the operator never
|
||||
// opted into auth. Pre-#49 the bearer was never inspected; this
|
||||
// preserves that for require_auth=false.
|
||||
Err(AuthError::InvalidKey) => {
|
||||
if fleet.require_auth {
|
||||
unauthorized("invalid API key")
|
||||
} else {
|
||||
|
||||
112
crates/cortex-gateway/src/entitlements_chain.rs
Normal file
112
crates/cortex-gateway/src/entitlements_chain.rs
Normal file
@@ -0,0 +1,112 @@
|
||||
//! Chained entitlement provider (#57): operator-local keys first, mesh
|
||||
//! upstream for everything else.
|
||||
//!
|
||||
//! `resolve` tries the [`LocalEntitlementProvider`] (operator + infra keys —
|
||||
//! never a network hop); only a locally-unknown key falls through to
|
||||
//! [`UpstreamEntitlementProvider`]. Because the local provider treats an
|
||||
//! unconfigured principal as uncapped, reserve/settle/release/snapshot must
|
||||
//! **not** blindly hit local — they dispatch to whichever backend resolved
|
||||
//! that account, remembered in a map keyed by `account_id` (populated at
|
||||
//! resolve time).
|
||||
|
||||
use crate::entitlements_local::LocalEntitlementProvider;
|
||||
use crate::entitlements_upstream::UpstreamEntitlementProvider;
|
||||
use async_trait::async_trait;
|
||||
use cortex_core::entitlements::{
|
||||
AuthError, BudgetError, BudgetSnapshot, EntitlementProvider, Principal, Reservation,
|
||||
};
|
||||
use std::collections::HashMap;
|
||||
use tokio::sync::RwLock;
|
||||
|
||||
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
|
||||
enum Backend {
|
||||
Local,
|
||||
Upstream,
|
||||
}
|
||||
|
||||
pub struct ChainedEntitlementProvider {
|
||||
local: LocalEntitlementProvider,
|
||||
upstream: UpstreamEntitlementProvider,
|
||||
/// account_id → which backend owns it, learned at resolve time.
|
||||
backends: RwLock<HashMap<String, Backend>>,
|
||||
}
|
||||
|
||||
impl ChainedEntitlementProvider {
|
||||
pub fn new(local: LocalEntitlementProvider, upstream: UpstreamEntitlementProvider) -> Self {
|
||||
Self {
|
||||
local,
|
||||
upstream,
|
||||
backends: RwLock::new(HashMap::new()),
|
||||
}
|
||||
}
|
||||
|
||||
async fn record(&self, account_id: &str, backend: Backend) {
|
||||
self.backends
|
||||
.write()
|
||||
.await
|
||||
.insert(account_id.to_string(), backend);
|
||||
}
|
||||
|
||||
/// The backend that owns `account_id`. Defaults to `Upstream` for an
|
||||
/// account never resolved this process-lifetime (a resolve always
|
||||
/// precedes reserve in a request, so this is just a safe fallback —
|
||||
/// upstream fails closed if the account is bogus).
|
||||
async fn backend_for(&self, account_id: &str) -> Backend {
|
||||
self.backends
|
||||
.read()
|
||||
.await
|
||||
.get(account_id)
|
||||
.copied()
|
||||
.unwrap_or(Backend::Upstream)
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl EntitlementProvider for ChainedEntitlementProvider {
|
||||
async fn resolve(&self, api_key: &str) -> Result<Principal, AuthError> {
|
||||
match self.local.resolve(api_key).await {
|
||||
Ok(p) => {
|
||||
self.record(&p.account_id, Backend::Local).await;
|
||||
Ok(p)
|
||||
}
|
||||
Err(AuthError::InvalidKey) => {
|
||||
let p = self.upstream.resolve(api_key).await?;
|
||||
self.record(&p.account_id, Backend::Upstream).await;
|
||||
Ok(p)
|
||||
}
|
||||
Err(e) => Err(e),
|
||||
}
|
||||
}
|
||||
|
||||
async fn reserve(
|
||||
&self,
|
||||
principal: &Principal,
|
||||
max_tokens: u64,
|
||||
) -> Result<Reservation, BudgetError> {
|
||||
match self.backend_for(&principal.account_id).await {
|
||||
Backend::Local => self.local.reserve(principal, max_tokens).await,
|
||||
Backend::Upstream => self.upstream.reserve(principal, max_tokens).await,
|
||||
}
|
||||
}
|
||||
|
||||
async fn settle(&self, reservation: Reservation, actual_tokens: u64) {
|
||||
match self.backend_for(&reservation.principal.account_id).await {
|
||||
Backend::Local => self.local.settle(reservation, actual_tokens).await,
|
||||
Backend::Upstream => self.upstream.settle(reservation, actual_tokens).await,
|
||||
}
|
||||
}
|
||||
|
||||
async fn release(&self, reservation: Reservation) {
|
||||
match self.backend_for(&reservation.principal.account_id).await {
|
||||
Backend::Local => self.local.release(reservation).await,
|
||||
Backend::Upstream => self.upstream.release(reservation).await,
|
||||
}
|
||||
}
|
||||
|
||||
async fn snapshot(&self, principal: &Principal) -> Option<BudgetSnapshot> {
|
||||
match self.backend_for(&principal.account_id).await {
|
||||
Backend::Local => self.local.snapshot(principal).await,
|
||||
Backend::Upstream => self.upstream.snapshot(principal).await,
|
||||
}
|
||||
}
|
||||
}
|
||||
246
crates/cortex-gateway/src/entitlements_upstream.rs
Normal file
246
crates/cortex-gateway/src/entitlements_upstream.rs
Normal file
@@ -0,0 +1,246 @@
|
||||
//! helexa-upstream client (#57): an [`EntitlementProvider`] that resolves
|
||||
//! keys and reserves/settles budget against the mesh authority's
|
||||
//! `/authz/v1` surface (B2). It is "just another impl of the trait" — cortex
|
||||
//! enforcement (`auth.rs`, `metering.rs`) is unchanged.
|
||||
//!
|
||||
//! **Fail closed.** When upstream is unreachable, `resolve` returns
|
||||
//! [`AuthError::Unavailable`] (→ `503`, never `401`) and `reserve` refuses
|
||||
//! with a retryable [`BudgetError::RateLimited`] — a request is never served
|
||||
//! on an un-authorized key, and a real key is never rejected as invalid
|
||||
//! during a blip.
|
||||
|
||||
use async_trait::async_trait;
|
||||
use cortex_core::config::UpstreamClientConfig;
|
||||
use cortex_core::entitlements::{
|
||||
AuthError, BudgetError, BudgetSnapshot, EntitlementProvider, Principal, Reservation,
|
||||
};
|
||||
use serde::Deserialize;
|
||||
use std::time::Duration;
|
||||
|
||||
/// Retry-After (seconds) advertised when we fail closed on an upstream
|
||||
/// outage.
|
||||
const FAIL_CLOSED_RETRY_SECS: u64 = 5;
|
||||
|
||||
pub struct UpstreamEntitlementProvider {
|
||||
client: reqwest::Client,
|
||||
base_url: String,
|
||||
bearer: String,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct PrincipalDto {
|
||||
account_id: String,
|
||||
key_id: String,
|
||||
}
|
||||
#[derive(Deserialize)]
|
||||
struct SnapshotDto {
|
||||
hard_cap: Option<u64>,
|
||||
spent: u64,
|
||||
reserved: u64,
|
||||
}
|
||||
#[derive(Deserialize)]
|
||||
struct ResolveResp {
|
||||
principal: PrincipalDto,
|
||||
#[allow(dead_code)]
|
||||
snapshot: Option<SnapshotDto>,
|
||||
}
|
||||
#[derive(Deserialize)]
|
||||
struct ReserveResp {
|
||||
reservation_id: Option<i64>,
|
||||
rejected: Option<Rejection>,
|
||||
}
|
||||
#[derive(Deserialize)]
|
||||
#[serde(tag = "kind", rename_all = "snake_case")]
|
||||
enum Rejection {
|
||||
InsufficientQuota {
|
||||
requested: u64,
|
||||
available: u64,
|
||||
},
|
||||
RateLimited {
|
||||
requested: u64,
|
||||
available: u64,
|
||||
retry_after_secs: u64,
|
||||
},
|
||||
}
|
||||
|
||||
impl UpstreamEntitlementProvider {
|
||||
pub fn new(cfg: &UpstreamClientConfig) -> Self {
|
||||
let client = reqwest::Client::builder()
|
||||
.timeout(Duration::from_secs(cfg.timeout_secs))
|
||||
.build()
|
||||
.expect("failed to build upstream HTTP client");
|
||||
Self {
|
||||
client,
|
||||
base_url: cfg.url.trim_end_matches('/').to_string(),
|
||||
bearer: cfg.bearer.clone(),
|
||||
}
|
||||
}
|
||||
|
||||
fn url(&self, path: &str) -> String {
|
||||
format!("{}{}", self.base_url, path)
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl EntitlementProvider for UpstreamEntitlementProvider {
|
||||
async fn resolve(&self, api_key: &str) -> Result<Principal, AuthError> {
|
||||
let resp = self
|
||||
.client
|
||||
.post(self.url("/authz/v1/resolve"))
|
||||
.bearer_auth(&self.bearer)
|
||||
.json(&serde_json::json!({ "api_key": api_key }))
|
||||
.send()
|
||||
.await;
|
||||
let resp = match resp {
|
||||
Ok(r) => r,
|
||||
Err(e) => {
|
||||
tracing::warn!(error = %e, "upstream resolve unreachable; failing closed");
|
||||
return Err(AuthError::Unavailable {
|
||||
retry_after_secs: FAIL_CLOSED_RETRY_SECS,
|
||||
});
|
||||
}
|
||||
};
|
||||
if resp.status().as_u16() == 401 {
|
||||
return Err(AuthError::InvalidKey);
|
||||
}
|
||||
if !resp.status().is_success() {
|
||||
return Err(AuthError::Unavailable {
|
||||
retry_after_secs: FAIL_CLOSED_RETRY_SECS,
|
||||
});
|
||||
}
|
||||
match resp.json::<ResolveResp>().await {
|
||||
Ok(r) => Ok(Principal {
|
||||
account_id: r.principal.account_id,
|
||||
key_id: r.principal.key_id,
|
||||
}),
|
||||
Err(e) => {
|
||||
tracing::warn!(error = %e, "upstream resolve: bad body; failing closed");
|
||||
Err(AuthError::Unavailable {
|
||||
retry_after_secs: FAIL_CLOSED_RETRY_SECS,
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn reserve(
|
||||
&self,
|
||||
principal: &Principal,
|
||||
max_tokens: u64,
|
||||
) -> Result<Reservation, BudgetError> {
|
||||
let fail_closed = || BudgetError::RateLimited {
|
||||
requested: max_tokens,
|
||||
available: 0,
|
||||
retry_after_secs: FAIL_CLOSED_RETRY_SECS,
|
||||
};
|
||||
let resp = self
|
||||
.client
|
||||
.post(self.url("/authz/v1/reserve"))
|
||||
.bearer_auth(&self.bearer)
|
||||
.json(&serde_json::json!({
|
||||
"account_id": principal.account_id,
|
||||
"key_id": principal.key_id,
|
||||
"max_tokens": max_tokens,
|
||||
}))
|
||||
.send()
|
||||
.await;
|
||||
let resp = match resp {
|
||||
Ok(r) if r.status().is_success() => r,
|
||||
Ok(r) => {
|
||||
tracing::warn!(status = %r.status(), "upstream reserve non-2xx; failing closed");
|
||||
return Err(fail_closed());
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::warn!(error = %e, "upstream reserve unreachable; failing closed");
|
||||
return Err(fail_closed());
|
||||
}
|
||||
};
|
||||
match resp.json::<ReserveResp>().await {
|
||||
Ok(ReserveResp {
|
||||
reservation_id: Some(id),
|
||||
..
|
||||
}) => Ok(Reservation {
|
||||
id: id as u64,
|
||||
principal: principal.clone(),
|
||||
reserved: max_tokens,
|
||||
}),
|
||||
Ok(ReserveResp {
|
||||
rejected:
|
||||
Some(Rejection::InsufficientQuota {
|
||||
requested,
|
||||
available,
|
||||
}),
|
||||
..
|
||||
}) => Err(BudgetError::InsufficientQuota {
|
||||
requested,
|
||||
available,
|
||||
}),
|
||||
Ok(ReserveResp {
|
||||
rejected:
|
||||
Some(Rejection::RateLimited {
|
||||
requested,
|
||||
available,
|
||||
retry_after_secs,
|
||||
}),
|
||||
..
|
||||
}) => Err(BudgetError::RateLimited {
|
||||
requested,
|
||||
available,
|
||||
retry_after_secs,
|
||||
}),
|
||||
_ => Err(fail_closed()),
|
||||
}
|
||||
}
|
||||
|
||||
async fn settle(&self, reservation: Reservation, actual_tokens: u64) {
|
||||
// Best-effort; a lost settle is reaped by the upstream sweeper (B2).
|
||||
let _ = self
|
||||
.client
|
||||
.post(self.url("/authz/v1/settle"))
|
||||
.bearer_auth(&self.bearer)
|
||||
.json(&serde_json::json!({
|
||||
"reservation_id": reservation.id as i64,
|
||||
"actual_tokens": actual_tokens,
|
||||
}))
|
||||
.send()
|
||||
.await
|
||||
.inspect_err(
|
||||
|e| tracing::warn!(error = %e, "upstream settle failed (sweeper will reap)"),
|
||||
);
|
||||
}
|
||||
|
||||
async fn release(&self, reservation: Reservation) {
|
||||
let _ = self
|
||||
.client
|
||||
.post(self.url("/authz/v1/release"))
|
||||
.bearer_auth(&self.bearer)
|
||||
.json(&serde_json::json!({ "reservation_id": reservation.id as i64 }))
|
||||
.send()
|
||||
.await
|
||||
.inspect_err(
|
||||
|e| tracing::warn!(error = %e, "upstream release failed (sweeper will reap)"),
|
||||
);
|
||||
}
|
||||
|
||||
async fn snapshot(&self, principal: &Principal) -> Option<BudgetSnapshot> {
|
||||
let resp = self
|
||||
.client
|
||||
.post(self.url("/authz/v1/snapshot"))
|
||||
.bearer_auth(&self.bearer)
|
||||
.json(&serde_json::json!({
|
||||
"account_id": principal.account_id,
|
||||
"key_id": principal.key_id,
|
||||
}))
|
||||
.send()
|
||||
.await
|
||||
.ok()?;
|
||||
if !resp.status().is_success() {
|
||||
return None;
|
||||
}
|
||||
let dto = resp.json::<SnapshotDto>().await.ok()?;
|
||||
Some(BudgetSnapshot {
|
||||
hard_cap: dto.hard_cap,
|
||||
spent: dto.spent,
|
||||
reserved: dto.reserved,
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -1,6 +1,8 @@
|
||||
pub mod anthropic_sse;
|
||||
pub mod auth;
|
||||
pub mod entitlements_chain;
|
||||
pub mod entitlements_local;
|
||||
pub mod entitlements_upstream;
|
||||
pub mod error;
|
||||
pub mod evictor;
|
||||
pub mod handlers;
|
||||
|
||||
@@ -1,4 +1,6 @@
|
||||
use crate::entitlements_chain::ChainedEntitlementProvider;
|
||||
use crate::entitlements_local::LocalEntitlementProvider;
|
||||
use crate::entitlements_upstream::UpstreamEntitlementProvider;
|
||||
use cortex_core::catalogue::ModelCatalogue;
|
||||
use cortex_core::config::{EvictionSettings, GatewayConfig, NeuronEndpoint};
|
||||
use cortex_core::entitlements::EntitlementProvider;
|
||||
@@ -45,8 +47,20 @@ impl CortexState {
|
||||
|
||||
let catalogue = ModelCatalogue::load(&config.models_config);
|
||||
|
||||
let entitlements: Arc<dyn EntitlementProvider> =
|
||||
Arc::new(LocalEntitlementProvider::from_config(&config.entitlements));
|
||||
// Local provider always handles operator + infra keys. When the
|
||||
// upstream client is enabled (#57), wrap it in the chain so locally
|
||||
// unknown keys fall through to the mesh authority; otherwise stay
|
||||
// purely local.
|
||||
let local = LocalEntitlementProvider::from_config(&config.entitlements);
|
||||
let entitlements: Arc<dyn EntitlementProvider> = if config.upstream.enabled {
|
||||
tracing::info!(url = %config.upstream.url, "upstream entitlement client enabled");
|
||||
Arc::new(ChainedEntitlementProvider::new(
|
||||
local,
|
||||
UpstreamEntitlementProvider::new(&config.upstream),
|
||||
))
|
||||
} else {
|
||||
Arc::new(local)
|
||||
};
|
||||
|
||||
Self {
|
||||
nodes: RwLock::new(nodes),
|
||||
|
||||
@@ -57,6 +57,7 @@ async fn test_alias_resolves_in_chat_completions() {
|
||||
}],
|
||||
models_config: models_path.to_string_lossy().to_string(),
|
||||
entitlements: Default::default(),
|
||||
upstream: Default::default(),
|
||||
};
|
||||
|
||||
let fleet = Arc::new(CortexState::from_config(&config));
|
||||
@@ -143,6 +144,7 @@ async fn test_aliases_surface_in_v1_models() {
|
||||
}],
|
||||
models_config: models_path.to_string_lossy().to_string(),
|
||||
entitlements: Default::default(),
|
||||
upstream: Default::default(),
|
||||
};
|
||||
|
||||
let fleet = Arc::new(CortexState::from_config(&config));
|
||||
@@ -232,6 +234,7 @@ async fn test_alias_falls_through_for_unmapped_model() {
|
||||
}],
|
||||
models_config: models_path.to_string_lossy().to_string(),
|
||||
entitlements: Default::default(),
|
||||
upstream: Default::default(),
|
||||
};
|
||||
|
||||
let fleet = Arc::new(CortexState::from_config(&config));
|
||||
|
||||
@@ -105,6 +105,7 @@ async fn spawn_gateway(neuron_url: &str, entitlements: EntitlementsConfig) -> St
|
||||
}],
|
||||
models_config: "/dev/null".into(),
|
||||
entitlements,
|
||||
upstream: Default::default(),
|
||||
};
|
||||
|
||||
let fleet = Arc::new(CortexState::from_config(&config));
|
||||
|
||||
@@ -81,6 +81,7 @@ async fn spawn_gateway(neuron_url: &str, key: ApiKeyConfig) -> (Arc<CortexState>
|
||||
require_auth: true,
|
||||
keys: vec![key],
|
||||
},
|
||||
upstream: Default::default(),
|
||||
};
|
||||
let fleet = Arc::new(CortexState::from_config(&config));
|
||||
{
|
||||
|
||||
@@ -430,6 +430,7 @@ pub async fn spawn_gateway_with_state(mock_url: &str) -> (Arc<CortexState>, Stri
|
||||
}],
|
||||
models_config: "/dev/null".into(),
|
||||
entitlements: Default::default(),
|
||||
upstream: Default::default(),
|
||||
};
|
||||
|
||||
let fleet = Arc::new(CortexState::from_config(&config));
|
||||
|
||||
@@ -89,6 +89,7 @@ async fn error_response_no_healthy_nodes() {
|
||||
}],
|
||||
models_config: "/dev/null".into(),
|
||||
entitlements: Default::default(),
|
||||
upstream: Default::default(),
|
||||
};
|
||||
|
||||
let fleet = Arc::new(cortex_gateway::state::CortexState::from_config(&config));
|
||||
|
||||
@@ -72,6 +72,7 @@ fn make_fleet(endpoint: &str, defrag_after: u32) -> Arc<CortexState> {
|
||||
}],
|
||||
models_config: "/dev/null".into(),
|
||||
entitlements: Default::default(),
|
||||
upstream: Default::default(),
|
||||
};
|
||||
Arc::new(CortexState::from_config(&config))
|
||||
}
|
||||
|
||||
@@ -74,6 +74,7 @@ async fn fleet_with(big_healthy: bool, big_devices: usize) -> Arc<CortexState> {
|
||||
],
|
||||
models_config: cat.to_string_lossy().into_owned(),
|
||||
entitlements: Default::default(),
|
||||
upstream: Default::default(),
|
||||
};
|
||||
let fleet = Arc::new(CortexState::from_config(&config));
|
||||
{
|
||||
|
||||
@@ -72,6 +72,7 @@ async fn two_neuron_fleet(endpoint_a: &str, endpoint_b: &str) -> Arc<CortexState
|
||||
],
|
||||
models_config: "/dev/null".into(),
|
||||
entitlements: Default::default(),
|
||||
upstream: Default::default(),
|
||||
};
|
||||
Arc::new(CortexState::from_config(&config))
|
||||
}
|
||||
|
||||
@@ -53,6 +53,7 @@ async fn spawn_metered_gateway(neuron_url: &str) -> (Arc<CortexState>, String) {
|
||||
window: CapWindow::Balance,
|
||||
}],
|
||||
},
|
||||
upstream: Default::default(),
|
||||
};
|
||||
|
||||
let fleet = Arc::new(CortexState::from_config(&config));
|
||||
@@ -158,6 +159,7 @@ async fn anonymous_request_records_no_spend() {
|
||||
}],
|
||||
models_config: "/dev/null".into(),
|
||||
entitlements: EntitlementsConfig::default(),
|
||||
upstream: Default::default(),
|
||||
};
|
||||
let fleet = Arc::new(CortexState::from_config(&config));
|
||||
{
|
||||
|
||||
@@ -66,6 +66,7 @@ harness = "candle"
|
||||
}],
|
||||
models_config: cat_path.to_string_lossy().into_owned(),
|
||||
entitlements: Default::default(),
|
||||
upstream: Default::default(),
|
||||
};
|
||||
|
||||
let fleet = Arc::new(CortexState::from_config(&config));
|
||||
|
||||
@@ -55,6 +55,7 @@ capabilities = ["text"]
|
||||
}],
|
||||
models_config: cat_path.to_string_lossy().into_owned(),
|
||||
entitlements: Default::default(),
|
||||
upstream: Default::default(),
|
||||
};
|
||||
|
||||
let fleet = Arc::new(CortexState::from_config(&config));
|
||||
|
||||
@@ -32,6 +32,7 @@ async fn test_poller_discovers_models() {
|
||||
}],
|
||||
models_config: "/dev/null".into(),
|
||||
entitlements: Default::default(),
|
||||
upstream: Default::default(),
|
||||
};
|
||||
|
||||
let fleet = Arc::new(CortexState::from_config(&config));
|
||||
@@ -84,6 +85,7 @@ async fn test_poller_updates_gateway_models_endpoint() {
|
||||
}],
|
||||
models_config: "/dev/null".into(),
|
||||
entitlements: Default::default(),
|
||||
upstream: Default::default(),
|
||||
};
|
||||
|
||||
let fleet = Arc::new(CortexState::from_config(&config));
|
||||
@@ -156,6 +158,7 @@ async fn test_models_endpoint_unions_capabilities_across_nodes() {
|
||||
],
|
||||
models_config: "/dev/null".into(),
|
||||
entitlements: Default::default(),
|
||||
upstream: Default::default(),
|
||||
};
|
||||
|
||||
let fleet = Arc::new(CortexState::from_config(&config));
|
||||
@@ -219,6 +222,7 @@ async fn test_poller_marks_unreachable_node_unhealthy() {
|
||||
}],
|
||||
models_config: "/dev/null".into(),
|
||||
entitlements: Default::default(),
|
||||
upstream: Default::default(),
|
||||
};
|
||||
|
||||
let fleet = Arc::new(CortexState::from_config(&config));
|
||||
@@ -273,6 +277,7 @@ async fn test_poller_removes_stale_models() {
|
||||
}],
|
||||
models_config: "/dev/null".into(),
|
||||
entitlements: Default::default(),
|
||||
upstream: Default::default(),
|
||||
};
|
||||
|
||||
let fleet = Arc::new(CortexState::from_config(&config));
|
||||
@@ -304,6 +309,7 @@ async fn test_poller_removes_stale_models() {
|
||||
}],
|
||||
models_config: "/dev/null".into(),
|
||||
entitlements: Default::default(),
|
||||
upstream: Default::default(),
|
||||
};
|
||||
|
||||
let fleet2 = Arc::new(CortexState::from_config(&config2));
|
||||
@@ -386,6 +392,7 @@ async fn test_poller_captures_activation_from_health() {
|
||||
}],
|
||||
models_config: "/dev/null".into(),
|
||||
entitlements: Default::default(),
|
||||
upstream: Default::default(),
|
||||
};
|
||||
|
||||
let fleet = Arc::new(CortexState::from_config(&config));
|
||||
@@ -431,6 +438,7 @@ async fn test_poller_parses_recovering_status() {
|
||||
}],
|
||||
models_config: "/dev/null".into(),
|
||||
entitlements: Default::default(),
|
||||
upstream: Default::default(),
|
||||
};
|
||||
|
||||
let fleet = Arc::new(CortexState::from_config(&config));
|
||||
|
||||
@@ -75,6 +75,7 @@ async fn spawn_gateway(neuron: &str, context: usize) -> String {
|
||||
}],
|
||||
models_config: "/dev/null".into(),
|
||||
entitlements: Default::default(),
|
||||
upstream: Default::default(),
|
||||
};
|
||||
let fleet = Arc::new(CortexState::from_config(&config));
|
||||
{
|
||||
|
||||
@@ -118,6 +118,7 @@ async fn test_no_healthy_nodes() {
|
||||
}],
|
||||
models_config: "/dev/null".into(),
|
||||
entitlements: Default::default(),
|
||||
upstream: Default::default(),
|
||||
};
|
||||
let fleet = std::sync::Arc::new(cortex_gateway::state::CortexState::from_config(&config));
|
||||
|
||||
|
||||
105
crates/cortex-gateway/tests/upstream_chain.rs
Normal file
105
crates/cortex-gateway/tests/upstream_chain.rs
Normal file
@@ -0,0 +1,105 @@
|
||||
//! B3: the chained entitlement provider (local → upstream) and fail-closed
|
||||
//! semantics, exercised against a mock helexa-upstream `/authz/v1`.
|
||||
|
||||
use axum::{Json, Router, routing::post};
|
||||
use cortex_core::config::{ApiKeyConfig, EntitlementsConfig, UpstreamClientConfig};
|
||||
use cortex_core::entitlements::{AuthError, EntitlementProvider};
|
||||
use cortex_gateway::entitlements_chain::ChainedEntitlementProvider;
|
||||
use cortex_gateway::entitlements_local::LocalEntitlementProvider;
|
||||
use cortex_gateway::entitlements_upstream::UpstreamEntitlementProvider;
|
||||
use serde_json::{Value, json};
|
||||
use tokio::net::TcpListener;
|
||||
|
||||
/// Mock upstream: `mesh-key` resolves to a mesh account; anything else 401.
|
||||
/// reserve always grants reservation 1.
|
||||
async fn spawn_mock_upstream() -> String {
|
||||
async fn resolve(Json(body): Json<Value>) -> axum::response::Response {
|
||||
use axum::response::IntoResponse;
|
||||
if body["api_key"] == "mesh-key" {
|
||||
Json(json!({"principal": {"account_id": "mesh-acct", "key_id": "mesh-key-1"}}))
|
||||
.into_response()
|
||||
} else {
|
||||
(
|
||||
axum::http::StatusCode::UNAUTHORIZED,
|
||||
Json(json!({"error": {"code": "invalid_api_key"}})),
|
||||
)
|
||||
.into_response()
|
||||
}
|
||||
}
|
||||
async fn reserve() -> Json<Value> {
|
||||
Json(json!({ "reservation_id": 1 }))
|
||||
}
|
||||
let app = Router::new()
|
||||
.route("/authz/v1/resolve", post(resolve))
|
||||
.route("/authz/v1/reserve", post(reserve));
|
||||
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||||
let addr = listener.local_addr().unwrap();
|
||||
tokio::spawn(async move {
|
||||
axum::serve(listener, app).await.unwrap();
|
||||
});
|
||||
format!("http://{addr}")
|
||||
}
|
||||
|
||||
fn local_with_key() -> LocalEntitlementProvider {
|
||||
let cfg = EntitlementsConfig {
|
||||
require_auth: false,
|
||||
keys: vec![ApiKeyConfig {
|
||||
key: "local-key".into(),
|
||||
account_id: "op".into(),
|
||||
key_id: None,
|
||||
hard_cap: None,
|
||||
window: Default::default(),
|
||||
}],
|
||||
};
|
||||
LocalEntitlementProvider::from_config(&cfg)
|
||||
}
|
||||
|
||||
fn chain(local: LocalEntitlementProvider, url: &str) -> ChainedEntitlementProvider {
|
||||
let upstream = UpstreamEntitlementProvider::new(&UpstreamClientConfig {
|
||||
enabled: true,
|
||||
url: url.to_string(),
|
||||
bearer: "client-secret".into(),
|
||||
timeout_secs: 5,
|
||||
});
|
||||
ChainedEntitlementProvider::new(local, upstream)
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn local_key_resolves_locally() {
|
||||
let url = spawn_mock_upstream().await;
|
||||
let c = chain(local_with_key(), &url);
|
||||
let p = c.resolve("local-key").await.expect("local resolves");
|
||||
assert_eq!(p.account_id, "op");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn unknown_key_falls_through_to_upstream() {
|
||||
let url = spawn_mock_upstream().await;
|
||||
let c = chain(local_with_key(), &url);
|
||||
let p = c.resolve("mesh-key").await.expect("upstream resolves");
|
||||
assert_eq!(p.account_id, "mesh-acct");
|
||||
assert_eq!(p.key_id, "mesh-key-1");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn unknown_everywhere_is_invalid_key() {
|
||||
let url = spawn_mock_upstream().await;
|
||||
let c = chain(local_with_key(), &url);
|
||||
match c.resolve("nope").await {
|
||||
Err(AuthError::InvalidKey) => {}
|
||||
other => panic!("expected InvalidKey, got {other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn upstream_unreachable_fails_closed_as_unavailable() {
|
||||
// No mock — point at a dead port. A locally-unknown key must surface
|
||||
// Unavailable (→ 503), never InvalidKey (→ 401).
|
||||
let c = chain(local_with_key(), "http://127.0.0.1:1");
|
||||
match c.resolve("some-mesh-key").await {
|
||||
Err(AuthError::Unavailable { retry_after_secs }) => assert!(retry_after_secs > 0),
|
||||
other => panic!("expected Unavailable, got {other:?}"),
|
||||
}
|
||||
// A local key still resolves without touching upstream.
|
||||
assert_eq!(c.resolve("local-key").await.unwrap().account_id, "op");
|
||||
}
|
||||
@@ -44,16 +44,6 @@ sqlx = { version = "0.8", default-features = false, features = [
|
||||
uuid = { version = "1", features = ["v4", "serde"] }
|
||||
sha2 = "0.10"
|
||||
subtle = "2.6"
|
||||
# Web auth (B4): argon2id password hashing, JWT sessions, CSPRNG secrets,
|
||||
# transactional email.
|
||||
argon2 = "0.5"
|
||||
jsonwebtoken = "9"
|
||||
rand = "0.8"
|
||||
lettre = { version = "0.11", default-features = false, features = [
|
||||
"tokio1-rustls-tls",
|
||||
"smtp-transport",
|
||||
"builder",
|
||||
] }
|
||||
|
||||
# cortex-core for the shared #63 OpenAiError envelope on the authz surface.
|
||||
cortex-core = { workspace = true }
|
||||
|
||||
@@ -22,64 +22,6 @@ pub struct UpstreamConfig {
|
||||
pub client_auth: ClientAuthSettings,
|
||||
#[serde(default)]
|
||||
pub authz: AuthzSettings,
|
||||
#[serde(default)]
|
||||
pub auth: AuthSettings,
|
||||
#[serde(default)]
|
||||
pub email: EmailSettings,
|
||||
}
|
||||
|
||||
/// `[auth]` — web-session signing + token lifetimes (B4). Web sessions are
|
||||
/// JWTs, distinct from inference API keys.
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct AuthSettings {
|
||||
/// HMAC secret for signing session JWTs. MUST be overridden in prod
|
||||
/// (env `UPSTREAM_AUTH__JWT_SECRET`); the default is dev-only.
|
||||
#[serde(default = "default_jwt_secret")]
|
||||
pub jwt_secret: String,
|
||||
/// Session token lifetime (seconds).
|
||||
#[serde(default = "default_session_ttl")]
|
||||
pub session_ttl_secs: u64,
|
||||
/// Email verification / password-reset token lifetime (seconds).
|
||||
#[serde(default = "default_email_token_ttl")]
|
||||
pub email_token_ttl_secs: u64,
|
||||
/// Public base URL of the frontend, used to build verify/reset links.
|
||||
#[serde(default = "default_app_base_url")]
|
||||
pub app_base_url: String,
|
||||
}
|
||||
|
||||
impl Default for AuthSettings {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
jwt_secret: default_jwt_secret(),
|
||||
session_ttl_secs: default_session_ttl(),
|
||||
email_token_ttl_secs: default_email_token_ttl(),
|
||||
app_base_url: default_app_base_url(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// `[email]` — transactional email transport for verify/reset.
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct EmailSettings {
|
||||
/// `"log"` (dev — logs the link) or `"smtp"`.
|
||||
#[serde(default = "default_email_provider")]
|
||||
pub provider: String,
|
||||
/// SMTP relay URL (e.g. "smtp://user:pass@host:587") when provider=smtp.
|
||||
#[serde(default)]
|
||||
pub smtp_url: Option<String>,
|
||||
/// `From:` address.
|
||||
#[serde(default = "default_from_addr")]
|
||||
pub from_addr: String,
|
||||
}
|
||||
|
||||
impl Default for EmailSettings {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
provider: default_email_provider(),
|
||||
smtp_url: None,
|
||||
from_addr: default_from_addr(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// `[client_auth]` — credentials operators' cortexes present to `/authz/v1`.
|
||||
@@ -200,24 +142,6 @@ fn default_reservation_ttl() -> u64 {
|
||||
fn default_sweep_interval() -> u64 {
|
||||
60
|
||||
}
|
||||
fn default_jwt_secret() -> String {
|
||||
"dev-insecure-change-me".into()
|
||||
}
|
||||
fn default_session_ttl() -> u64 {
|
||||
7 * 24 * 3600
|
||||
}
|
||||
fn default_email_token_ttl() -> u64 {
|
||||
24 * 3600
|
||||
}
|
||||
fn default_app_base_url() -> String {
|
||||
"http://localhost:5173".into()
|
||||
}
|
||||
fn default_email_provider() -> String {
|
||||
"log".into()
|
||||
}
|
||||
fn default_from_addr() -> String {
|
||||
"helexa <no-reply@helexa.ai>".into()
|
||||
}
|
||||
|
||||
impl UpstreamConfig {
|
||||
/// Load from a TOML file with `UPSTREAM_`-prefixed env overrides
|
||||
|
||||
@@ -1,14 +1,7 @@
|
||||
//! Hashing + secret-generation helpers.
|
||||
//!
|
||||
//! - **Passwords** (low-entropy) → argon2id PHC strings.
|
||||
//! - **API keys / top-up codes / email + session tokens** (high-entropy
|
||||
//! secrets minted here) → stored only as their sha256; sha256 is the fast,
|
||||
//! sufficient choice for high-entropy material.
|
||||
//! Hashing helpers. API keys and top-up codes are stored only as their
|
||||
//! sha256 (they are high-entropy secrets; sha256 is the fast, sufficient
|
||||
//! choice — argon2 is reserved for low-entropy passwords).
|
||||
|
||||
use argon2::Argon2;
|
||||
use argon2::password_hash::rand_core::OsRng as ArgonOsRng;
|
||||
use argon2::password_hash::{PasswordHash, PasswordHasher, PasswordVerifier, SaltString};
|
||||
use rand::RngCore;
|
||||
use sha2::{Digest, Sha256};
|
||||
|
||||
/// sha256 of `input`, as raw bytes (matches the `BYTEA` columns).
|
||||
@@ -17,97 +10,3 @@ pub fn sha256(input: &str) -> Vec<u8> {
|
||||
h.update(input.as_bytes());
|
||||
h.finalize().to_vec()
|
||||
}
|
||||
|
||||
/// Hash a password with argon2id, returning a PHC string for storage.
|
||||
pub fn hash_password(password: &str) -> Result<String, argon2::password_hash::Error> {
|
||||
let salt = SaltString::generate(&mut ArgonOsRng);
|
||||
Ok(Argon2::default()
|
||||
.hash_password(password.as_bytes(), &salt)?
|
||||
.to_string())
|
||||
}
|
||||
|
||||
/// Verify a password against a stored PHC hash. `false` on any mismatch or
|
||||
/// malformed hash (never panics).
|
||||
pub fn verify_password(password: &str, phc: &str) -> bool {
|
||||
match PasswordHash::new(phc) {
|
||||
Ok(parsed) => Argon2::default()
|
||||
.verify_password(password.as_bytes(), &parsed)
|
||||
.is_ok(),
|
||||
Err(_) => false,
|
||||
}
|
||||
}
|
||||
|
||||
/// A fresh URL-safe high-entropy secret (256 bits) for email/session/reset
|
||||
/// tokens. The caller stores only `sha256` of this and emails/returns the
|
||||
/// raw value.
|
||||
pub fn random_token() -> String {
|
||||
let mut bytes = [0u8; 32];
|
||||
rand::rngs::OsRng.fill_bytes(&mut bytes);
|
||||
base62(&bytes)
|
||||
}
|
||||
|
||||
/// Mint a new API key: `(raw, prefix)`. `raw` is shown to the user once;
|
||||
/// only `sha256(raw)` is stored. The prefix is a non-secret display tag.
|
||||
pub fn generate_api_key() -> (String, String) {
|
||||
let mut bytes = [0u8; 32];
|
||||
rand::rngs::OsRng.fill_bytes(&mut bytes);
|
||||
let raw = format!("sk-helexa-{}", base62(&bytes));
|
||||
// Non-secret prefix for the dashboard list (scheme + first few chars).
|
||||
let prefix: String = raw.chars().take(14).collect();
|
||||
(raw, prefix)
|
||||
}
|
||||
|
||||
/// base62 encode (0-9A-Za-z) — URL/clipboard friendly, no padding.
|
||||
fn base62(bytes: &[u8]) -> String {
|
||||
const ALPHABET: &[u8] = b"0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz";
|
||||
// Treat the bytes as a big-endian integer and base62 it. 32 bytes → ~43
|
||||
// chars. Simple repeated-division over a big-uint built from the bytes.
|
||||
let mut digits: Vec<u8> = vec![0];
|
||||
for &byte in bytes {
|
||||
let mut carry = byte as u32;
|
||||
for d in digits.iter_mut() {
|
||||
let v = (*d as u32) * 256 + carry;
|
||||
*d = (v % 62) as u8;
|
||||
carry = v / 62;
|
||||
}
|
||||
while carry > 0 {
|
||||
digits.push((carry % 62) as u8);
|
||||
carry /= 62;
|
||||
}
|
||||
}
|
||||
digits
|
||||
.iter()
|
||||
.rev()
|
||||
.map(|&d| ALPHABET[d as usize] as char)
|
||||
.collect()
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn password_round_trips_and_rejects_wrong() {
|
||||
let phc = hash_password("correct horse").unwrap();
|
||||
assert!(verify_password("correct horse", &phc));
|
||||
assert!(!verify_password("wrong", &phc));
|
||||
assert!(!verify_password("correct horse", "not-a-phc-string"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn api_key_has_scheme_prefix_and_unique_body() {
|
||||
let (raw, prefix) = generate_api_key();
|
||||
assert!(raw.starts_with("sk-helexa-"));
|
||||
assert!(prefix.starts_with("sk-helexa-"));
|
||||
let (raw2, _) = generate_api_key();
|
||||
assert_ne!(raw, raw2, "keys are unique");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn random_tokens_are_unique_and_nonempty() {
|
||||
let a = random_token();
|
||||
let b = random_token();
|
||||
assert!(!a.is_empty());
|
||||
assert_ne!(a, b);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,64 +0,0 @@
|
||||
//! Transactional email for verification + password-reset links.
|
||||
//!
|
||||
//! Two transports: `Log` (dev — writes the link to the log so flows are
|
||||
//! testable without a relay) and `Smtp` (lettre over rustls). Built from
|
||||
//! `[email]` config.
|
||||
|
||||
use crate::config::EmailSettings;
|
||||
use anyhow::{Context, Result};
|
||||
use lettre::message::Mailbox;
|
||||
use lettre::{AsyncSmtpTransport, AsyncTransport, Message, Tokio1Executor};
|
||||
|
||||
#[derive(Clone)]
|
||||
pub enum EmailSender {
|
||||
/// Dev: log the message instead of sending.
|
||||
Log { from: String },
|
||||
Smtp {
|
||||
from: String,
|
||||
transport: AsyncSmtpTransport<Tokio1Executor>,
|
||||
},
|
||||
}
|
||||
|
||||
impl EmailSender {
|
||||
pub fn from_config(cfg: &EmailSettings) -> Result<Self> {
|
||||
match cfg.provider.as_str() {
|
||||
"smtp" => {
|
||||
let url = cfg
|
||||
.smtp_url
|
||||
.as_deref()
|
||||
.context("[email].smtp_url required when provider = \"smtp\"")?;
|
||||
let transport = AsyncSmtpTransport::<Tokio1Executor>::from_url(url)
|
||||
.context("parsing [email].smtp_url")?
|
||||
.build();
|
||||
Ok(EmailSender::Smtp {
|
||||
from: cfg.from_addr.clone(),
|
||||
transport,
|
||||
})
|
||||
}
|
||||
_ => Ok(EmailSender::Log {
|
||||
from: cfg.from_addr.clone(),
|
||||
}),
|
||||
}
|
||||
}
|
||||
|
||||
/// Send a plaintext email. Errors are returned but the caller treats
|
||||
/// send failures as non-fatal to the request (the user can re-request).
|
||||
pub async fn send(&self, to: &str, subject: &str, body: &str) -> Result<()> {
|
||||
match self {
|
||||
EmailSender::Log { from } => {
|
||||
tracing::info!(%from, %to, %subject, body, "EMAIL (log transport)");
|
||||
Ok(())
|
||||
}
|
||||
EmailSender::Smtp { from, transport } => {
|
||||
let msg = Message::builder()
|
||||
.from(from.parse::<Mailbox>().context("parsing from_addr")?)
|
||||
.to(to.parse::<Mailbox>().context("parsing recipient")?)
|
||||
.subject(subject)
|
||||
.body(body.to_string())
|
||||
.context("building message")?;
|
||||
transport.send(msg).await.context("sending email")?;
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -16,20 +16,15 @@ pub mod authz;
|
||||
pub mod config;
|
||||
pub mod crypto;
|
||||
pub mod db;
|
||||
pub mod email;
|
||||
pub mod error;
|
||||
pub mod handlers;
|
||||
pub mod ledger;
|
||||
pub mod state;
|
||||
pub mod topup;
|
||||
pub mod web;
|
||||
|
||||
use anyhow::Result;
|
||||
use config::UpstreamConfig;
|
||||
use email::EmailSender;
|
||||
use state::AppState;
|
||||
use std::time::Duration;
|
||||
use tower_http::cors::CorsLayer;
|
||||
use tower_http::trace::TraceLayer;
|
||||
|
||||
/// Build the axum application.
|
||||
@@ -37,11 +32,6 @@ pub fn build_app(state: AppState) -> axum::Router {
|
||||
axum::Router::new()
|
||||
.merge(handlers::routes())
|
||||
.merge(authz::router(&state))
|
||||
.merge(web::router(&state))
|
||||
// The /web/v1 surface is called cross-origin by the browser SPA in
|
||||
// dev; same-origin behind nginx in prod. Permissive is fine — these
|
||||
// endpoints authenticate via bearer/JWT, not cookies.
|
||||
.layer(CorsLayer::permissive())
|
||||
.layer(TraceLayer::new_for_http())
|
||||
.with_state(state)
|
||||
}
|
||||
@@ -50,9 +40,8 @@ pub fn build_app(state: AppState) -> axum::Router {
|
||||
/// reservation sweeper, bind the listener.
|
||||
pub async fn run(config: UpstreamConfig) -> Result<()> {
|
||||
let pool = db::connect_and_migrate(&config.db.url, config.db.max_connections).await?;
|
||||
let email = EmailSender::from_config(&config.email)?;
|
||||
let listen = config.server.listen.clone();
|
||||
let state = AppState::new(pool, config, email);
|
||||
let state = AppState::new(pool, config);
|
||||
|
||||
if state.config.client_auth.tokens.is_empty() {
|
||||
tracing::warn!(
|
||||
|
||||
@@ -20,22 +20,6 @@ enum Commands {
|
||||
#[arg(short, long, default_value = "helexa-upstream.toml")]
|
||||
config: String,
|
||||
},
|
||||
/// Mint single-use top-up codes and print them (one per line). The raw
|
||||
/// codes are shown only here — only their hash is stored. (The future
|
||||
/// faucet bot calls the same path.)
|
||||
Mint {
|
||||
#[arg(short, long, default_value = "helexa-upstream.toml")]
|
||||
config: String,
|
||||
/// Tokens each code grants.
|
||||
#[arg(long)]
|
||||
value: i64,
|
||||
/// How many codes to mint.
|
||||
#[arg(long, default_value_t = 1)]
|
||||
count: u32,
|
||||
/// Optional human label (e.g. "small", "beta-launch").
|
||||
#[arg(long)]
|
||||
denomination: Option<String>,
|
||||
},
|
||||
}
|
||||
|
||||
#[tokio::main]
|
||||
@@ -56,25 +40,6 @@ async fn main() -> Result<()> {
|
||||
tracing::info!(listen = %cfg.server.listen, "starting helexa-upstream");
|
||||
helexa_upstream::run(cfg).await?;
|
||||
}
|
||||
Commands::Mint {
|
||||
config,
|
||||
value,
|
||||
count,
|
||||
denomination,
|
||||
} => {
|
||||
let cfg = UpstreamConfig::load(&config)
|
||||
.map_err(|e| anyhow::anyhow!("failed to load config from '{config}': {e}"))?;
|
||||
let pool =
|
||||
helexa_upstream::db::connect_and_migrate(&cfg.db.url, cfg.db.max_connections)
|
||||
.await?;
|
||||
let codes =
|
||||
helexa_upstream::topup::mint(&pool, value, count, denomination.as_deref()).await?;
|
||||
// Raw codes to stdout (one per line) for the operator to distribute;
|
||||
// logs/diagnostics go to stderr via tracing.
|
||||
for code in codes {
|
||||
println!("{code}");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Ok(())
|
||||
|
||||
@@ -1,7 +1,6 @@
|
||||
//! Shared application state.
|
||||
|
||||
use crate::config::UpstreamConfig;
|
||||
use crate::email::EmailSender;
|
||||
use sqlx::postgres::PgPool;
|
||||
use std::sync::Arc;
|
||||
|
||||
@@ -9,15 +8,13 @@ use std::sync::Arc;
|
||||
pub struct AppState {
|
||||
pub pool: PgPool,
|
||||
pub config: Arc<UpstreamConfig>,
|
||||
pub email: EmailSender,
|
||||
}
|
||||
|
||||
impl AppState {
|
||||
pub fn new(pool: PgPool, config: UpstreamConfig, email: EmailSender) -> Self {
|
||||
pub fn new(pool: PgPool, config: UpstreamConfig) -> Self {
|
||||
Self {
|
||||
pool,
|
||||
config: Arc::new(config),
|
||||
email,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,82 +0,0 @@
|
||||
//! Single-use top-up codes (#B5) — the second half of the hybrid allocation
|
||||
//! model. Each code grants `value` tokens to the account that redeems it,
|
||||
//! raising `accounts.allocation_total`. Minting codes is operator/CLI side
|
||||
//! (the future faucet bot calls the same `mint` path); redemption is a
|
||||
//! `/web/v1` action.
|
||||
//!
|
||||
//! Security: only `sha256(code)` is stored. Redemption is **timing-safe and
|
||||
//! single-use** — a conditional `UPDATE … WHERE redeemed_by IS NULL` does
|
||||
//! the claim atomically (concurrent double-redeem → exactly one winner), and
|
||||
//! a not-found code and an already-redeemed code return the **same** generic
|
||||
//! failure with the same code path (no oracle for "valid but spent").
|
||||
|
||||
use crate::crypto::{random_token, sha256};
|
||||
use sqlx::Row;
|
||||
use sqlx::postgres::PgPool;
|
||||
use uuid::Uuid;
|
||||
|
||||
#[derive(Debug, thiserror::Error)]
|
||||
pub enum TopUpError {
|
||||
/// Code unknown OR already redeemed — deliberately indistinguishable.
|
||||
#[error("invalid or already-redeemed code")]
|
||||
Invalid,
|
||||
#[error(transparent)]
|
||||
Db(#[from] sqlx::Error),
|
||||
}
|
||||
|
||||
/// Redeem `raw_code` for `account_id`, raising the account's
|
||||
/// `allocation_total` by the code's value. Returns the new total.
|
||||
pub async fn redeem(pool: &PgPool, account_id: Uuid, raw_code: &str) -> Result<i64, TopUpError> {
|
||||
let mut tx = pool.begin().await?;
|
||||
// Atomic single-use claim. `redeemed_by IS NULL` is the guarantee: under
|
||||
// concurrent redemption exactly one UPDATE touches the row.
|
||||
let claimed = sqlx::query(
|
||||
"UPDATE top_up_codes SET redeemed_by = $1, redeemed_at = now() \
|
||||
WHERE code_hash = $2 AND redeemed_by IS NULL RETURNING value",
|
||||
)
|
||||
.bind(account_id)
|
||||
.bind(sha256(raw_code))
|
||||
.fetch_optional(&mut *tx)
|
||||
.await?;
|
||||
let Some(row) = claimed else {
|
||||
// Not found or already redeemed — same path, same error.
|
||||
return Err(TopUpError::Invalid);
|
||||
};
|
||||
let value: i64 = row.get("value");
|
||||
let new_total: i64 = sqlx::query(
|
||||
"UPDATE accounts SET allocation_total = allocation_total + $1 WHERE id = $2 \
|
||||
RETURNING allocation_total",
|
||||
)
|
||||
.bind(value)
|
||||
.bind(account_id)
|
||||
.fetch_one(&mut *tx)
|
||||
.await?
|
||||
.get("allocation_total");
|
||||
tx.commit().await?;
|
||||
Ok(new_total)
|
||||
}
|
||||
|
||||
/// Mint `count` codes each worth `value` tokens, optionally tagged with a
|
||||
/// `denomination` label. Returns the raw codes (shown once — only their
|
||||
/// hash is stored). The CLI prints these; the future faucet bot calls this.
|
||||
pub async fn mint(
|
||||
pool: &PgPool,
|
||||
value: i64,
|
||||
count: u32,
|
||||
denomination: Option<&str>,
|
||||
) -> Result<Vec<String>, sqlx::Error> {
|
||||
let mut codes = Vec::with_capacity(count as usize);
|
||||
for _ in 0..count {
|
||||
let raw = format!("helexa-topup-{}", random_token());
|
||||
sqlx::query(
|
||||
"INSERT INTO top_up_codes (code_hash, value, denomination) VALUES ($1, $2, $3)",
|
||||
)
|
||||
.bind(sha256(&raw))
|
||||
.bind(value)
|
||||
.bind(denomination)
|
||||
.execute(pool)
|
||||
.await?;
|
||||
codes.push(raw);
|
||||
}
|
||||
Ok(codes)
|
||||
}
|
||||
@@ -1,594 +0,0 @@
|
||||
//! `/web/v1` — the human-facing account API the helexa.ai frontend (#F4)
|
||||
//! consumes: email+password auth (register / verify / login / reset),
|
||||
//! API-key CRUD with per-key limits, and the account balance. Web sessions
|
||||
//! are JWTs, **distinct** from inference API keys.
|
||||
//!
|
||||
//! Errors use a plain JSON shape `{ "error": { "message", "code" } }` (web
|
||||
//! clients, not OpenAI clients — the #63 envelope is the authz surface).
|
||||
//!
|
||||
//! Silent fingerprint abuse (no clue to the abuser): registration captures
|
||||
//! the browser fingerprint and always succeeds; when ≥ threshold accounts
|
||||
//! share one fingerprint, all are silently `deactivated` (keys then resolve
|
||||
//! as ordinary `401`s at the authz surface — never a "banned" signal).
|
||||
|
||||
use crate::crypto::{generate_api_key, hash_password, random_token, sha256, verify_password};
|
||||
use crate::state::AppState;
|
||||
use axum::extract::{Path, Request, State};
|
||||
use axum::http::{StatusCode, header};
|
||||
use axum::middleware::Next;
|
||||
use axum::response::{IntoResponse, Json, Response};
|
||||
use axum::routing::{get, post};
|
||||
use axum::{Extension, Router};
|
||||
use chrono::{DateTime, Duration, Utc};
|
||||
use jsonwebtoken::{DecodingKey, EncodingKey, Header, Validation, decode, encode};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use serde_json::json;
|
||||
use sqlx::Row;
|
||||
use uuid::Uuid;
|
||||
|
||||
pub fn router(state: &AppState) -> Router<AppState> {
|
||||
let protected = Router::new()
|
||||
.route("/web/v1/account", get(account))
|
||||
.route("/web/v1/keys", get(list_keys).post(create_key))
|
||||
.route("/web/v1/keys/{id}/archive", post(archive_key))
|
||||
.route(
|
||||
"/web/v1/keys/{id}/limit",
|
||||
axum::routing::patch(update_key_limit),
|
||||
)
|
||||
.route("/web/v1/redeem", post(redeem))
|
||||
.layer(axum::middleware::from_fn_with_state(
|
||||
state.clone(),
|
||||
require_session,
|
||||
));
|
||||
|
||||
Router::new()
|
||||
.route("/web/v1/register", post(register))
|
||||
.route("/web/v1/verify", post(verify))
|
||||
.route("/web/v1/login", post(login))
|
||||
.route("/web/v1/password-reset/request", post(reset_request))
|
||||
.route("/web/v1/password-reset/confirm", post(reset_confirm))
|
||||
.merge(protected)
|
||||
}
|
||||
|
||||
// ── errors ──────────────────────────────────────────────────────────
|
||||
|
||||
enum WebError {
|
||||
BadRequest(&'static str),
|
||||
Unauthorized,
|
||||
Internal,
|
||||
}
|
||||
|
||||
impl IntoResponse for WebError {
|
||||
fn into_response(self) -> Response {
|
||||
let (status, code, message) = match self {
|
||||
WebError::BadRequest(m) => (StatusCode::BAD_REQUEST, "bad_request", m),
|
||||
WebError::Unauthorized => (StatusCode::UNAUTHORIZED, "unauthorized", "unauthorized"),
|
||||
WebError::Internal => (
|
||||
StatusCode::INTERNAL_SERVER_ERROR,
|
||||
"internal_error",
|
||||
"internal error",
|
||||
),
|
||||
};
|
||||
(
|
||||
status,
|
||||
Json(json!({"error": {"message": message, "code": code}})),
|
||||
)
|
||||
.into_response()
|
||||
}
|
||||
}
|
||||
|
||||
impl From<sqlx::Error> for WebError {
|
||||
fn from(e: sqlx::Error) -> Self {
|
||||
tracing::error!(error = %e, "web db error");
|
||||
WebError::Internal
|
||||
}
|
||||
}
|
||||
|
||||
type WebResult<T> = Result<T, WebError>;
|
||||
|
||||
// ── sessions (JWT) ──────────────────────────────────────────────────
|
||||
|
||||
#[derive(Serialize, Deserialize)]
|
||||
struct Claims {
|
||||
sub: String, // user id
|
||||
exp: usize,
|
||||
}
|
||||
|
||||
fn mint_session(state: &AppState, user_id: Uuid) -> WebResult<String> {
|
||||
let exp = (Utc::now() + Duration::seconds(state.config.auth.session_ttl_secs as i64))
|
||||
.timestamp() as usize;
|
||||
let claims = Claims {
|
||||
sub: user_id.to_string(),
|
||||
exp,
|
||||
};
|
||||
encode(
|
||||
&Header::default(),
|
||||
&claims,
|
||||
&EncodingKey::from_secret(state.config.auth.jwt_secret.as_bytes()),
|
||||
)
|
||||
.map_err(|_| WebError::Internal)
|
||||
}
|
||||
|
||||
/// Authenticated user id, injected by [`require_session`].
|
||||
#[derive(Clone)]
|
||||
struct AuthUser(Uuid);
|
||||
|
||||
async fn require_session(State(state): State<AppState>, mut req: Request, next: Next) -> Response {
|
||||
let token = req
|
||||
.headers()
|
||||
.get(header::AUTHORIZATION)
|
||||
.and_then(|v| v.to_str().ok())
|
||||
.and_then(|v| v.strip_prefix("Bearer "))
|
||||
.map(str::trim);
|
||||
let Some(token) = token else {
|
||||
return WebError::Unauthorized.into_response();
|
||||
};
|
||||
let decoded = decode::<Claims>(
|
||||
token,
|
||||
&DecodingKey::from_secret(state.config.auth.jwt_secret.as_bytes()),
|
||||
&Validation::default(),
|
||||
);
|
||||
match decoded
|
||||
.ok()
|
||||
.and_then(|d| Uuid::parse_str(&d.claims.sub).ok())
|
||||
{
|
||||
Some(uid) => {
|
||||
req.extensions_mut().insert(AuthUser(uid));
|
||||
next.run(req).await
|
||||
}
|
||||
None => WebError::Unauthorized.into_response(),
|
||||
}
|
||||
}
|
||||
|
||||
/// The caller's single account id.
|
||||
async fn account_id_for(state: &AppState, user_id: Uuid) -> WebResult<Uuid> {
|
||||
let row = sqlx::query("SELECT id FROM accounts WHERE owner_user_id = $1")
|
||||
.bind(user_id)
|
||||
.fetch_optional(&state.pool)
|
||||
.await?;
|
||||
row.map(|r| r.get::<Uuid, _>("id"))
|
||||
.ok_or(WebError::Internal)
|
||||
}
|
||||
|
||||
// ── auth lifecycle ──────────────────────────────────────────────────
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct RegisterReq {
|
||||
email: String,
|
||||
password: String,
|
||||
#[serde(default)]
|
||||
fingerprint: Option<String>,
|
||||
}
|
||||
|
||||
/// `POST /web/v1/register` — always returns `202`, regardless of whether the
|
||||
/// email was new, already taken, or fingerprint-flagged (no enumeration, no
|
||||
/// abuse clue).
|
||||
async fn register(State(state): State<AppState>, Json(req): Json<RegisterReq>) -> Response {
|
||||
match register_inner(&state, req).await {
|
||||
Ok(()) | Err(WebError::BadRequest(_)) => {}
|
||||
Err(e) => return e.into_response(),
|
||||
}
|
||||
// Generic 202 whatever happened above (except hard server errors).
|
||||
StatusCode::ACCEPTED.into_response()
|
||||
}
|
||||
|
||||
async fn register_inner(state: &AppState, req: RegisterReq) -> WebResult<()> {
|
||||
if !req.email.contains('@') {
|
||||
return Err(WebError::BadRequest("invalid email"));
|
||||
}
|
||||
if req.password.len() < 8 {
|
||||
return Err(WebError::BadRequest("password too short (min 8)"));
|
||||
}
|
||||
let phc = hash_password(&req.password).map_err(|_| WebError::Internal)?;
|
||||
|
||||
// Insert the user; a duplicate email silently no-ops (no enumeration).
|
||||
let user_id: Option<Uuid> = sqlx::query(
|
||||
"INSERT INTO users (email, password_hash, registration_fingerprint) \
|
||||
VALUES ($1, $2, $3) ON CONFLICT (email) DO NOTHING RETURNING id",
|
||||
)
|
||||
.bind(&req.email)
|
||||
.bind(&phc)
|
||||
.bind(&req.fingerprint)
|
||||
.fetch_optional(&state.pool)
|
||||
.await?
|
||||
.map(|r| r.get("id"));
|
||||
|
||||
let Some(user_id) = user_id else {
|
||||
return Ok(()); // email already registered — say nothing
|
||||
};
|
||||
|
||||
// Account with the flat free grant.
|
||||
sqlx::query("INSERT INTO accounts (owner_user_id, allocation_total) VALUES ($1, $2)")
|
||||
.bind(user_id)
|
||||
.bind(state.config.grant.free_token_grant)
|
||||
.execute(&state.pool)
|
||||
.await?;
|
||||
|
||||
// Silent fingerprint abuse handling.
|
||||
if let Some(fp) = req.fingerprint.as_deref().filter(|f| !f.is_empty()) {
|
||||
apply_fingerprint_policy(state, fp).await?;
|
||||
}
|
||||
|
||||
// Email verification link.
|
||||
let token = random_token();
|
||||
let expires: DateTime<Utc> =
|
||||
Utc::now() + Duration::seconds(state.config.auth.email_token_ttl_secs as i64);
|
||||
sqlx::query(
|
||||
"INSERT INTO email_tokens (token_hash, user_id, kind, expires_at) \
|
||||
VALUES ($1, $2, 'verify', $3)",
|
||||
)
|
||||
.bind(sha256(&token))
|
||||
.bind(user_id)
|
||||
.bind(expires)
|
||||
.execute(&state.pool)
|
||||
.await?;
|
||||
|
||||
let link = format!("{}/verify?token={token}", state.config.auth.app_base_url);
|
||||
let _ = state
|
||||
.email
|
||||
.send(
|
||||
&req.email,
|
||||
"Verify your helexa account",
|
||||
&format!("Welcome to helexa. Verify your email:\n\n{link}\n"),
|
||||
)
|
||||
.await;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Count accounts sharing `fp`; flag them, and silently deactivate all once
|
||||
/// the count reaches the configured threshold. No response difference — the
|
||||
/// abuser gets no signal.
|
||||
async fn apply_fingerprint_policy(state: &AppState, fp: &str) -> WebResult<()> {
|
||||
let count: i64 =
|
||||
sqlx::query_scalar("SELECT count(*) FROM users WHERE registration_fingerprint = $1")
|
||||
.bind(fp)
|
||||
.fetch_one(&state.pool)
|
||||
.await?;
|
||||
if count > 1 {
|
||||
sqlx::query(
|
||||
"UPDATE accounts SET fingerprint_flagged = true \
|
||||
WHERE owner_user_id IN (SELECT id FROM users WHERE registration_fingerprint = $1)",
|
||||
)
|
||||
.bind(fp)
|
||||
.execute(&state.pool)
|
||||
.await?;
|
||||
}
|
||||
if count >= state.config.abuse.fingerprint_account_threshold {
|
||||
let res = sqlx::query(
|
||||
"UPDATE accounts SET status = 'deactivated' \
|
||||
WHERE owner_user_id IN (SELECT id FROM users WHERE registration_fingerprint = $1)",
|
||||
)
|
||||
.bind(fp)
|
||||
.execute(&state.pool)
|
||||
.await?;
|
||||
tracing::warn!(
|
||||
fingerprint = fp,
|
||||
accounts = res.rows_affected(),
|
||||
"silently deactivated fingerprint-abusing accounts"
|
||||
);
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct TokenReq {
|
||||
token: String,
|
||||
}
|
||||
|
||||
/// `POST /web/v1/verify` — consume a verification token, mark verified.
|
||||
async fn verify(State(state): State<AppState>, Json(req): Json<TokenReq>) -> WebResult<Response> {
|
||||
let row = sqlx::query(
|
||||
"UPDATE email_tokens SET consumed_at = now() \
|
||||
WHERE token_hash = $1 AND kind = 'verify' AND consumed_at IS NULL AND expires_at > now() \
|
||||
RETURNING user_id",
|
||||
)
|
||||
.bind(sha256(&req.token))
|
||||
.fetch_optional(&state.pool)
|
||||
.await?;
|
||||
let Some(row) = row else {
|
||||
return Err(WebError::BadRequest("invalid or expired token"));
|
||||
};
|
||||
let user_id: Uuid = row.get("user_id");
|
||||
sqlx::query("UPDATE users SET email_verified = true WHERE id = $1")
|
||||
.bind(user_id)
|
||||
.execute(&state.pool)
|
||||
.await?;
|
||||
Ok(StatusCode::OK.into_response())
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct LoginReq {
|
||||
email: String,
|
||||
password: String,
|
||||
}
|
||||
|
||||
/// `POST /web/v1/login` — verify password + email-verified → session JWT.
|
||||
async fn login(State(state): State<AppState>, Json(req): Json<LoginReq>) -> WebResult<Response> {
|
||||
let row = sqlx::query("SELECT id, password_hash, email_verified FROM users WHERE email = $1")
|
||||
.bind(&req.email)
|
||||
.fetch_optional(&state.pool)
|
||||
.await?;
|
||||
// Generic 401 for every failure mode (no enumeration).
|
||||
let Some(row) = row else {
|
||||
return Err(WebError::Unauthorized);
|
||||
};
|
||||
let phc: String = row.get("password_hash");
|
||||
let verified: bool = row.get("email_verified");
|
||||
if !verify_password(&req.password, &phc) || !verified {
|
||||
return Err(WebError::Unauthorized);
|
||||
}
|
||||
let user_id: Uuid = row.get("id");
|
||||
let token = mint_session(&state, user_id)?;
|
||||
Ok(Json(json!({
|
||||
"token": token,
|
||||
"expires_in": state.config.auth.session_ttl_secs,
|
||||
}))
|
||||
.into_response())
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct EmailReq {
|
||||
email: String,
|
||||
}
|
||||
|
||||
/// `POST /web/v1/password-reset/request` — always `202` (no enumeration);
|
||||
/// mints + emails a reset token only if the account exists.
|
||||
async fn reset_request(State(state): State<AppState>, Json(req): Json<EmailReq>) -> Response {
|
||||
// The inner only ever yields `Internal` (DB failure); a missing email is
|
||||
// Ok(()) so there's no enumeration. Surface 500 on a real error, else 202.
|
||||
match reset_request_inner(&state, &req.email).await {
|
||||
Ok(()) => StatusCode::ACCEPTED.into_response(),
|
||||
Err(e) => e.into_response(),
|
||||
}
|
||||
}
|
||||
|
||||
async fn reset_request_inner(state: &AppState, email: &str) -> WebResult<()> {
|
||||
let row = sqlx::query("SELECT id FROM users WHERE email = $1")
|
||||
.bind(email)
|
||||
.fetch_optional(&state.pool)
|
||||
.await?;
|
||||
let Some(row) = row else { return Ok(()) };
|
||||
let user_id: Uuid = row.get("id");
|
||||
let token = random_token();
|
||||
let expires: DateTime<Utc> =
|
||||
Utc::now() + Duration::seconds(state.config.auth.email_token_ttl_secs as i64);
|
||||
sqlx::query(
|
||||
"INSERT INTO email_tokens (token_hash, user_id, kind, expires_at) \
|
||||
VALUES ($1, $2, 'reset', $3)",
|
||||
)
|
||||
.bind(sha256(&token))
|
||||
.bind(user_id)
|
||||
.bind(expires)
|
||||
.execute(&state.pool)
|
||||
.await?;
|
||||
let link = format!("{}/reset?token={token}", state.config.auth.app_base_url);
|
||||
let _ = state
|
||||
.email
|
||||
.send(
|
||||
email,
|
||||
"Reset your helexa password",
|
||||
&format!("Reset your password:\n\n{link}\n"),
|
||||
)
|
||||
.await;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct ResetConfirmReq {
|
||||
token: String,
|
||||
new_password: String,
|
||||
}
|
||||
|
||||
/// `POST /web/v1/password-reset/confirm` — consume reset token, rotate hash.
|
||||
async fn reset_confirm(
|
||||
State(state): State<AppState>,
|
||||
Json(req): Json<ResetConfirmReq>,
|
||||
) -> WebResult<Response> {
|
||||
if req.new_password.len() < 8 {
|
||||
return Err(WebError::BadRequest("password too short (min 8)"));
|
||||
}
|
||||
let row = sqlx::query(
|
||||
"UPDATE email_tokens SET consumed_at = now() \
|
||||
WHERE token_hash = $1 AND kind = 'reset' AND consumed_at IS NULL AND expires_at > now() \
|
||||
RETURNING user_id",
|
||||
)
|
||||
.bind(sha256(&req.token))
|
||||
.fetch_optional(&state.pool)
|
||||
.await?;
|
||||
let Some(row) = row else {
|
||||
return Err(WebError::BadRequest("invalid or expired token"));
|
||||
};
|
||||
let user_id: Uuid = row.get("user_id");
|
||||
let phc = hash_password(&req.new_password).map_err(|_| WebError::Internal)?;
|
||||
sqlx::query("UPDATE users SET password_hash = $1 WHERE id = $2")
|
||||
.bind(phc)
|
||||
.bind(user_id)
|
||||
.execute(&state.pool)
|
||||
.await?;
|
||||
Ok(StatusCode::OK.into_response())
|
||||
}
|
||||
|
||||
// ── account + keys (protected) ──────────────────────────────────────
|
||||
|
||||
async fn account(
|
||||
State(state): State<AppState>,
|
||||
Extension(user): Extension<AuthUser>,
|
||||
) -> WebResult<Response> {
|
||||
let acct = account_id_for(&state, user.0).await?;
|
||||
let row = sqlx::query(
|
||||
"SELECT allocation_total, allocation_spent, allocation_reserved FROM accounts WHERE id = $1",
|
||||
)
|
||||
.bind(acct)
|
||||
.fetch_one(&state.pool)
|
||||
.await?;
|
||||
Ok(Json(json!({
|
||||
"account_id": acct.to_string(),
|
||||
"allocation_total": row.get::<i64, _>("allocation_total"),
|
||||
"allocation_spent": row.get::<i64, _>("allocation_spent"),
|
||||
"allocation_reserved": row.get::<i64, _>("allocation_reserved"),
|
||||
}))
|
||||
.into_response())
|
||||
}
|
||||
|
||||
async fn list_keys(
|
||||
State(state): State<AppState>,
|
||||
Extension(user): Extension<AuthUser>,
|
||||
) -> WebResult<Response> {
|
||||
let acct = account_id_for(&state, user.0).await?;
|
||||
let rows = sqlx::query(
|
||||
"SELECT id, key_prefix, label, status, limit_kind, limit_value, key_spent, key_reserved, \
|
||||
created_at \
|
||||
FROM api_keys WHERE account_id = $1 ORDER BY created_at DESC",
|
||||
)
|
||||
.bind(acct)
|
||||
.fetch_all(&state.pool)
|
||||
.await?;
|
||||
let keys: Vec<_> = rows
|
||||
.iter()
|
||||
.map(|r| {
|
||||
json!({
|
||||
"id": r.get::<Uuid, _>("id").to_string(),
|
||||
"prefix": r.get::<String, _>("key_prefix"),
|
||||
"label": r.get::<String, _>("label"),
|
||||
"status": r.get::<String, _>("status"),
|
||||
"limit_kind": r.get::<String, _>("limit_kind"),
|
||||
"limit_value": r.get::<i64, _>("limit_value"),
|
||||
"spent": r.get::<i64, _>("key_spent"),
|
||||
"reserved": r.get::<i64, _>("key_reserved"),
|
||||
"created_at": r.get::<DateTime<Utc>, _>("created_at").to_rfc3339(),
|
||||
})
|
||||
})
|
||||
.collect();
|
||||
Ok(Json(json!({ "keys": keys })).into_response())
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct CreateKeyReq {
|
||||
#[serde(default)]
|
||||
label: String,
|
||||
/// "percent" | "hardcap" (default percent=100 → full allocation).
|
||||
#[serde(default)]
|
||||
limit_kind: Option<String>,
|
||||
#[serde(default)]
|
||||
limit_value: Option<i64>,
|
||||
}
|
||||
|
||||
async fn create_key(
|
||||
State(state): State<AppState>,
|
||||
Extension(user): Extension<AuthUser>,
|
||||
Json(req): Json<CreateKeyReq>,
|
||||
) -> WebResult<Response> {
|
||||
let acct = account_id_for(&state, user.0).await?;
|
||||
let limit_kind = match req.limit_kind.as_deref() {
|
||||
Some("hardcap") => "hardcap",
|
||||
_ => "percent",
|
||||
};
|
||||
let limit_value = req.limit_value.unwrap_or(100).max(0);
|
||||
let (raw, prefix) = generate_api_key();
|
||||
let id: Uuid = sqlx::query(
|
||||
"INSERT INTO api_keys (account_id, key_hash, key_prefix, label, limit_kind, limit_value) \
|
||||
VALUES ($1, $2, $3, $4, $5, $6) RETURNING id",
|
||||
)
|
||||
.bind(acct)
|
||||
.bind(sha256(&raw))
|
||||
.bind(&prefix)
|
||||
.bind(&req.label)
|
||||
.bind(limit_kind)
|
||||
.bind(limit_value)
|
||||
.fetch_one(&state.pool)
|
||||
.await?
|
||||
.get("id");
|
||||
// The raw key is shown exactly once.
|
||||
Ok((
|
||||
StatusCode::CREATED,
|
||||
Json(json!({
|
||||
"id": id.to_string(),
|
||||
"key": raw,
|
||||
"prefix": prefix,
|
||||
"limit_kind": limit_kind,
|
||||
"limit_value": limit_value,
|
||||
})),
|
||||
)
|
||||
.into_response())
|
||||
}
|
||||
|
||||
async fn archive_key(
|
||||
State(state): State<AppState>,
|
||||
Extension(user): Extension<AuthUser>,
|
||||
Path(id): Path<Uuid>,
|
||||
) -> WebResult<Response> {
|
||||
let acct = account_id_for(&state, user.0).await?;
|
||||
let res = sqlx::query(
|
||||
"UPDATE api_keys SET status = 'archived' WHERE id = $1 AND account_id = $2 AND status = 'active'",
|
||||
)
|
||||
.bind(id)
|
||||
.bind(acct)
|
||||
.execute(&state.pool)
|
||||
.await?;
|
||||
if res.rows_affected() == 0 {
|
||||
return Err(WebError::BadRequest("no such active key"));
|
||||
}
|
||||
Ok(StatusCode::NO_CONTENT.into_response())
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct UpdateLimitReq {
|
||||
limit_kind: String,
|
||||
limit_value: i64,
|
||||
}
|
||||
|
||||
async fn update_key_limit(
|
||||
State(state): State<AppState>,
|
||||
Extension(user): Extension<AuthUser>,
|
||||
Path(id): Path<Uuid>,
|
||||
Json(req): Json<UpdateLimitReq>,
|
||||
) -> WebResult<Response> {
|
||||
if req.limit_kind != "percent" && req.limit_kind != "hardcap" {
|
||||
return Err(WebError::BadRequest(
|
||||
"limit_kind must be percent or hardcap",
|
||||
));
|
||||
}
|
||||
if req.limit_value < 0 {
|
||||
return Err(WebError::BadRequest("limit_value must be >= 0"));
|
||||
}
|
||||
let acct = account_id_for(&state, user.0).await?;
|
||||
let res = sqlx::query(
|
||||
"UPDATE api_keys SET limit_kind = $1, limit_value = $2 WHERE id = $3 AND account_id = $4",
|
||||
)
|
||||
.bind(&req.limit_kind)
|
||||
.bind(req.limit_value)
|
||||
.bind(id)
|
||||
.bind(acct)
|
||||
.execute(&state.pool)
|
||||
.await?;
|
||||
if res.rows_affected() == 0 {
|
||||
return Err(WebError::BadRequest("no such key"));
|
||||
}
|
||||
Ok(StatusCode::NO_CONTENT.into_response())
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct RedeemReq {
|
||||
code: String,
|
||||
}
|
||||
|
||||
/// `POST /web/v1/redeem` — redeem a single-use top-up code, raising the
|
||||
/// account's allocation. Returns the new total. Generic 400 for an invalid
|
||||
/// or already-redeemed code (no oracle).
|
||||
async fn redeem(
|
||||
State(state): State<AppState>,
|
||||
Extension(user): Extension<AuthUser>,
|
||||
Json(req): Json<RedeemReq>,
|
||||
) -> WebResult<Response> {
|
||||
let acct = account_id_for(&state, user.0).await?;
|
||||
match crate::topup::redeem(&state.pool, acct, &req.code).await {
|
||||
Ok(new_total) => Ok(Json(json!({ "allocation_total": new_total })).into_response()),
|
||||
Err(crate::topup::TopUpError::Invalid) => {
|
||||
Err(WebError::BadRequest("invalid or already-redeemed code"))
|
||||
}
|
||||
Err(crate::topup::TopUpError::Db(e)) => {
|
||||
tracing::error!(error = %e, "redeem db error");
|
||||
Err(WebError::Internal)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -34,16 +34,13 @@ async fn spawn_or_skip(test: &str) -> Option<(String, PgPool)> {
|
||||
abuse: Default::default(),
|
||||
client_auth: Default::default(),
|
||||
authz: Default::default(),
|
||||
auth: Default::default(),
|
||||
email: Default::default(),
|
||||
};
|
||||
config.client_auth.tokens.push(ClientToken {
|
||||
token: CLIENT_TOKEN.into(),
|
||||
operator_id: "op-test".into(),
|
||||
});
|
||||
|
||||
let email = helexa_upstream::email::EmailSender::from_config(&config.email).unwrap();
|
||||
let state = AppState::new(pool.clone(), config, email);
|
||||
let state = AppState::new(pool.clone(), config);
|
||||
let app = helexa_upstream::build_app(state);
|
||||
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||||
let addr = listener.local_addr().unwrap();
|
||||
|
||||
@@ -1,425 +0,0 @@
|
||||
//! Integration tests for the `/web/v1` account API + the silent fingerprint
|
||||
//! abuse policy, driving the built app over HTTP against a real Postgres.
|
||||
//! Gated on `UPSTREAM_TEST_DATABASE_URL` (skips cleanly when unset).
|
||||
|
||||
use helexa_upstream::config::{ClientToken, UpstreamConfig};
|
||||
use helexa_upstream::crypto::sha256;
|
||||
use helexa_upstream::db::connect_and_migrate;
|
||||
use helexa_upstream::email::EmailSender;
|
||||
use helexa_upstream::state::AppState;
|
||||
use serde_json::{Value, json};
|
||||
use sqlx::Executor;
|
||||
use sqlx::Row;
|
||||
use sqlx::postgres::PgPool;
|
||||
|
||||
const CLIENT_TOKEN: &str = "web-test-operator-token";
|
||||
|
||||
async fn spawn_or_skip(test: &str) -> Option<(String, PgPool)> {
|
||||
let Ok(url) = std::env::var("UPSTREAM_TEST_DATABASE_URL") else {
|
||||
eprintln!("skipping {test}: UPSTREAM_TEST_DATABASE_URL not set");
|
||||
return None;
|
||||
};
|
||||
let pool = connect_and_migrate(&url, 16).await.expect("migrate");
|
||||
let mut config = UpstreamConfig {
|
||||
server: Default::default(),
|
||||
db: helexa_upstream::config::DbSettings {
|
||||
url,
|
||||
max_connections: 16,
|
||||
},
|
||||
grant: Default::default(),
|
||||
abuse: Default::default(),
|
||||
client_auth: Default::default(),
|
||||
authz: Default::default(),
|
||||
auth: Default::default(),
|
||||
email: Default::default(), // Log transport
|
||||
};
|
||||
config.client_auth.tokens.push(ClientToken {
|
||||
token: CLIENT_TOKEN.into(),
|
||||
operator_id: "op-web".into(),
|
||||
});
|
||||
let email = EmailSender::from_config(&config.email).unwrap();
|
||||
let state = AppState::new(pool.clone(), config, email);
|
||||
let app = helexa_upstream::build_app(state);
|
||||
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||||
let addr = listener.local_addr().unwrap();
|
||||
tokio::spawn(async move {
|
||||
axum::serve(listener, app).await.unwrap();
|
||||
});
|
||||
Some((format!("http://{addr}"), pool))
|
||||
}
|
||||
|
||||
fn unique_email() -> String {
|
||||
format!("u-{}@test.local", uuid::Uuid::new_v4())
|
||||
}
|
||||
|
||||
async fn post(url: String, body: Value, bearer: Option<&str>) -> reqwest::Response {
|
||||
let c = reqwest::Client::new();
|
||||
let mut req = c.post(url).json(&body);
|
||||
if let Some(b) = bearer {
|
||||
req = req.bearer_auth(b);
|
||||
}
|
||||
req.send().await.unwrap()
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn verify_endpoint_consumes_token_once() {
|
||||
let Some((base, pool)) = spawn_or_skip("verify_endpoint_consumes_token_once").await else {
|
||||
return;
|
||||
};
|
||||
let email = unique_email();
|
||||
// Register, then mint a verify token directly (the raw token is only in
|
||||
// the email; here we insert a known one to drive the endpoint).
|
||||
assert_eq!(
|
||||
post(
|
||||
format!("{base}/web/v1/register"),
|
||||
json!({"email": email, "password": "password123"}),
|
||||
None
|
||||
)
|
||||
.await
|
||||
.status(),
|
||||
202
|
||||
);
|
||||
let user_id: uuid::Uuid = pool
|
||||
.fetch_one(sqlx::query("SELECT id FROM users WHERE email = $1").bind(&email))
|
||||
.await
|
||||
.unwrap()
|
||||
.get("id");
|
||||
let raw = "verify-raw-token-xyz";
|
||||
pool.execute(
|
||||
sqlx::query(
|
||||
"INSERT INTO email_tokens (token_hash, user_id, kind, expires_at) \
|
||||
VALUES ($1, $2, 'verify', now() + interval '1 hour')",
|
||||
)
|
||||
.bind(sha256(raw))
|
||||
.bind(user_id),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(
|
||||
post(format!("{base}/web/v1/verify"), json!({"token": raw}), None)
|
||||
.await
|
||||
.status(),
|
||||
200
|
||||
);
|
||||
// Consumed → second attempt fails.
|
||||
assert_eq!(
|
||||
post(format!("{base}/web/v1/verify"), json!({"token": raw}), None)
|
||||
.await
|
||||
.status(),
|
||||
400
|
||||
);
|
||||
|
||||
let verified: bool = pool
|
||||
.fetch_one(sqlx::query("SELECT email_verified FROM users WHERE id = $1").bind(user_id))
|
||||
.await
|
||||
.unwrap()
|
||||
.get("email_verified");
|
||||
assert!(verified);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn account_lifecycle_and_key_resolves_then_archives() {
|
||||
let Some((base, pool)) =
|
||||
spawn_or_skip("account_lifecycle_and_key_resolves_then_archives").await
|
||||
else {
|
||||
return;
|
||||
};
|
||||
let email = unique_email();
|
||||
post(
|
||||
format!("{base}/web/v1/register"),
|
||||
json!({"email": email, "password": "password123"}),
|
||||
None,
|
||||
)
|
||||
.await;
|
||||
// Bypass the email step for the login/key portion.
|
||||
pool.execute(
|
||||
sqlx::query("UPDATE users SET email_verified = true WHERE email = $1").bind(&email),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
// login → session JWT
|
||||
let r = post(
|
||||
format!("{base}/web/v1/login"),
|
||||
json!({"email": email, "password": "password123"}),
|
||||
None,
|
||||
)
|
||||
.await;
|
||||
assert_eq!(r.status(), 200);
|
||||
let token = r.json::<Value>().await.unwrap()["token"]
|
||||
.as_str()
|
||||
.unwrap()
|
||||
.to_string();
|
||||
|
||||
// create key (raw shown once)
|
||||
let r = post(
|
||||
format!("{base}/web/v1/keys"),
|
||||
json!({"label": "laptop"}),
|
||||
Some(&token),
|
||||
)
|
||||
.await;
|
||||
assert_eq!(r.status(), 201);
|
||||
let body: Value = r.json().await.unwrap();
|
||||
let raw_key = body["key"].as_str().unwrap().to_string();
|
||||
let key_id = body["id"].as_str().unwrap().to_string();
|
||||
assert!(raw_key.starts_with("sk-helexa-"));
|
||||
|
||||
// account balance reflects the free grant
|
||||
let r = reqwest::Client::new()
|
||||
.get(format!("{base}/web/v1/account"))
|
||||
.bearer_auth(&token)
|
||||
.send()
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
r.json::<Value>().await.unwrap()["allocation_total"],
|
||||
1_000_000
|
||||
);
|
||||
|
||||
// list keys shows the prefix, never the raw secret
|
||||
let r = reqwest::Client::new()
|
||||
.get(format!("{base}/web/v1/keys"))
|
||||
.bearer_auth(&token)
|
||||
.send()
|
||||
.await
|
||||
.unwrap();
|
||||
let listed = r.json::<Value>().await.unwrap();
|
||||
let k = &listed["keys"][0];
|
||||
assert_eq!(k["id"], key_id);
|
||||
assert!(k.get("key").is_none(), "raw secret never listed");
|
||||
assert!(k["prefix"].as_str().unwrap().starts_with("sk-helexa-"));
|
||||
|
||||
// the key authorizes at the authz surface
|
||||
let r = post(
|
||||
format!("{base}/authz/v1/resolve"),
|
||||
json!({"api_key": raw_key}),
|
||||
Some(CLIENT_TOKEN),
|
||||
)
|
||||
.await;
|
||||
assert_eq!(r.status(), 200);
|
||||
|
||||
// archive → the key no longer resolves
|
||||
let r = post(
|
||||
format!("{base}/web/v1/keys/{key_id}/archive"),
|
||||
json!({}),
|
||||
Some(&token),
|
||||
)
|
||||
.await;
|
||||
assert_eq!(r.status(), 204);
|
||||
let r = post(
|
||||
format!("{base}/authz/v1/resolve"),
|
||||
json!({"api_key": raw_key}),
|
||||
Some(CLIENT_TOKEN),
|
||||
)
|
||||
.await;
|
||||
assert_eq!(r.status(), 401);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn fingerprint_abuse_silently_deactivates_all_no_clue() {
|
||||
let Some((base, pool)) =
|
||||
spawn_or_skip("fingerprint_abuse_silently_deactivates_all_no_clue").await
|
||||
else {
|
||||
return;
|
||||
};
|
||||
let fp = format!("fp-{}", uuid::Uuid::new_v4());
|
||||
|
||||
// 5 registrations sharing one fingerprint — every one returns a normal 202.
|
||||
let mut emails = Vec::new();
|
||||
for _ in 0..5 {
|
||||
let email = unique_email();
|
||||
let r = post(
|
||||
format!("{base}/web/v1/register"),
|
||||
json!({"email": email, "password": "password123", "fingerprint": fp}),
|
||||
None,
|
||||
)
|
||||
.await;
|
||||
assert_eq!(r.status(), 202, "registration always looks successful");
|
||||
emails.push(email);
|
||||
}
|
||||
|
||||
// Silent effect: all 5 accounts are deactivated + flagged.
|
||||
let (deactivated, flagged): (i64, i64) = {
|
||||
let row = pool
|
||||
.fetch_one(
|
||||
sqlx::query(
|
||||
"SELECT \
|
||||
count(*) FILTER (WHERE a.status = 'deactivated') AS d, \
|
||||
count(*) FILTER (WHERE a.fingerprint_flagged) AS f \
|
||||
FROM accounts a JOIN users u ON u.id = a.owner_user_id \
|
||||
WHERE u.registration_fingerprint = $1",
|
||||
)
|
||||
.bind(&fp),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
(row.get("d"), row.get("f"))
|
||||
};
|
||||
assert_eq!(deactivated, 5, "all sharing accounts silently deactivated");
|
||||
assert_eq!(flagged, 5);
|
||||
|
||||
// No clue at the authz surface: a key on a deactivated account resolves
|
||||
// as an ordinary 401, indistinguishable from an unknown key.
|
||||
let acct: uuid::Uuid = pool
|
||||
.fetch_one(
|
||||
sqlx::query(
|
||||
"SELECT a.id FROM accounts a JOIN users u ON u.id = a.owner_user_id \
|
||||
WHERE u.registration_fingerprint = $1 LIMIT 1",
|
||||
)
|
||||
.bind(&fp),
|
||||
)
|
||||
.await
|
||||
.unwrap()
|
||||
.get("id");
|
||||
let raw = "sk-helexa-deactivated-probe";
|
||||
pool.execute(
|
||||
sqlx::query(
|
||||
"INSERT INTO api_keys (account_id, key_hash, key_prefix) VALUES ($1, $2, 'sk-helexa-')",
|
||||
)
|
||||
.bind(acct)
|
||||
.bind(sha256(raw)),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
let r = post(
|
||||
format!("{base}/authz/v1/resolve"),
|
||||
json!({"api_key": raw}),
|
||||
Some(CLIENT_TOKEN),
|
||||
)
|
||||
.await;
|
||||
assert_eq!(
|
||||
r.status(),
|
||||
401,
|
||||
"deactivated account's key looks like any invalid key"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn topup_redeem_raises_allocation_single_use() {
|
||||
let Some((base, pool)) = spawn_or_skip("topup_redeem_raises_allocation_single_use").await
|
||||
else {
|
||||
return;
|
||||
};
|
||||
let email = unique_email();
|
||||
post(
|
||||
format!("{base}/web/v1/register"),
|
||||
json!({"email": email, "password": "password123"}),
|
||||
None,
|
||||
)
|
||||
.await;
|
||||
pool.execute(
|
||||
sqlx::query("UPDATE users SET email_verified = true WHERE email = $1").bind(&email),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
let token = post(
|
||||
format!("{base}/web/v1/login"),
|
||||
json!({"email": email, "password": "password123"}),
|
||||
None,
|
||||
)
|
||||
.await
|
||||
.json::<Value>()
|
||||
.await
|
||||
.unwrap()["token"]
|
||||
.as_str()
|
||||
.unwrap()
|
||||
.to_string();
|
||||
|
||||
// Mint a code worth 500_000 (mint path used by the CLI/faucet).
|
||||
let codes = helexa_upstream::topup::mint(&pool, 500_000, 1, Some("test"))
|
||||
.await
|
||||
.unwrap();
|
||||
let code = &codes[0];
|
||||
|
||||
// Redeem → allocation_total rises from the 1_000_000 free grant.
|
||||
let r = post(
|
||||
format!("{base}/web/v1/redeem"),
|
||||
json!({"code": code}),
|
||||
Some(&token),
|
||||
)
|
||||
.await;
|
||||
assert_eq!(r.status(), 200);
|
||||
assert_eq!(
|
||||
r.json::<Value>().await.unwrap()["allocation_total"],
|
||||
1_500_000
|
||||
);
|
||||
|
||||
// Single-use: a second redemption fails generically (no oracle).
|
||||
let r = post(
|
||||
format!("{base}/web/v1/redeem"),
|
||||
json!({"code": code}),
|
||||
Some(&token),
|
||||
)
|
||||
.await;
|
||||
assert_eq!(r.status(), 400);
|
||||
|
||||
// Unknown code: same generic 400.
|
||||
let r = post(
|
||||
format!("{base}/web/v1/redeem"),
|
||||
json!({"code": "helexa-topup-does-not-exist"}),
|
||||
Some(&token),
|
||||
)
|
||||
.await;
|
||||
assert_eq!(r.status(), 400);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn topup_concurrent_double_redeem_one_winner() {
|
||||
let Some((base, pool)) = spawn_or_skip("topup_concurrent_double_redeem_one_winner").await
|
||||
else {
|
||||
return;
|
||||
};
|
||||
// Two verified accounts.
|
||||
let mut tokens = Vec::new();
|
||||
for _ in 0..2 {
|
||||
let email = unique_email();
|
||||
post(
|
||||
format!("{base}/web/v1/register"),
|
||||
json!({"email": email, "password": "password123"}),
|
||||
None,
|
||||
)
|
||||
.await;
|
||||
pool.execute(
|
||||
sqlx::query("UPDATE users SET email_verified = true WHERE email = $1").bind(&email),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
let t = post(
|
||||
format!("{base}/web/v1/login"),
|
||||
json!({"email": email, "password": "password123"}),
|
||||
None,
|
||||
)
|
||||
.await
|
||||
.json::<Value>()
|
||||
.await
|
||||
.unwrap()["token"]
|
||||
.as_str()
|
||||
.unwrap()
|
||||
.to_string();
|
||||
tokens.push(t);
|
||||
}
|
||||
let code = helexa_upstream::topup::mint(&pool, 100, 1, None)
|
||||
.await
|
||||
.unwrap()
|
||||
.remove(0);
|
||||
|
||||
// Both accounts race to redeem the same code; exactly one wins.
|
||||
let (a, b) = tokio::join!(
|
||||
post(
|
||||
format!("{base}/web/v1/redeem"),
|
||||
json!({"code": code}),
|
||||
Some(&tokens[0])
|
||||
),
|
||||
post(
|
||||
format!("{base}/web/v1/redeem"),
|
||||
json!({"code": code}),
|
||||
Some(&tokens[1])
|
||||
),
|
||||
);
|
||||
let wins = [a.status(), b.status()]
|
||||
.iter()
|
||||
.filter(|s| s.as_u16() == 200)
|
||||
.count();
|
||||
assert_eq!(wins, 1, "exactly one redemption wins the single-use code");
|
||||
}
|
||||
@@ -35,18 +35,3 @@ url = "postgres://helexa:helexa@localhost/helexa_upstream"
|
||||
# reservation whose settle/release from a cortex was lost.
|
||||
# reservation_ttl_secs = 120
|
||||
# sweep_interval_secs = 60
|
||||
|
||||
[auth]
|
||||
# HMAC secret for signing web-session JWTs. MUST be overridden in prod via
|
||||
# UPSTREAM_AUTH__JWT_SECRET; the built-in default is dev-only.
|
||||
# jwt_secret = "change-me"
|
||||
# session_ttl_secs = 604800 # 7 days
|
||||
# email_token_ttl_secs = 86400 # 24 hours
|
||||
# Frontend base URL used to build verify/reset links in emails.
|
||||
app_base_url = "https://helexa.ai"
|
||||
|
||||
[email]
|
||||
# "log" (dev: logs the link) or "smtp".
|
||||
provider = "log"
|
||||
# smtp_url = "smtp://user:pass@smtp.example.com:587"
|
||||
from_addr = "helexa <no-reply@helexa.ai>"
|
||||
|
||||
Reference in New Issue
Block a user