refactor: consolidate duplicated DTO types and token entities

Author: evansims

Cargo.lock

@@ -80,7 +80,7 @@ aes-gcm = "0.10"
 argon2 = { version = "0.5", features = ["std"] }
 base64 = "0.22"
 chrono = { version = "0.4", features = ["serde"] }
-ed25519-dalek = { version = "2", features = ["rand_core", "zeroize"] }
+ed25519-dalek = { version = "2", features = ["rand_core", "zeroize", "pkcs8"] }
 hex = "0.4"
 jsonwebtoken = { version = "10", default-features = false, features = [
   "aws_lc_rs",
@@ -101,6 +101,9 @@ webauthn-rs = { version = "0.5", features = [
 # Snowflake ID generation
 idgenerator = "2"
 
+# Testing
+proptest = "1"
+
 # Email
 lettre = { version = "0.11", default-features = false, features = [
   "builder",
@@ -141,6 +144,9 @@ kube = { version = "3", features = ["client"] }
 # Caching
 moka = { version = "0.12", features = ["sync"] }
 
+# NTP client
+rsntp = { version = "4", default-features = false, features = ["chrono", "async"] }
+
 # TLS (for crypto provider initialization)
 rustls = { version = "0.23", default-features = false, features = [
   "aws_lc_rs",

Cargo.toml

@@ -43,7 +43,9 @@ uuid = { workspace = true }
 [dev-dependencies]
 inferadb-control-test-fixtures = { path = "../test-fixtures" }
 inferadb-common-authn = ">=0.1.0-0"
+proptest = { workspace = true }
 rand_core = { version = "0.6", features = ["getrandom"] }
+sha2 = { workspace = true }
 
 [lints]
 workspace = true

MANIFEST.md

@@ -38,6 +38,8 @@ pub struct AppState {
     pub leader: Option<Arc<inferadb_control_core::LeaderElection<Backend>>>,
     pub email_service: Option<Arc<inferadb_control_core::EmailService>>,
     pub control_identity: Option<Arc<inferadb_control_types::ControlIdentity>>,
+    #[builder(default)]
+    pub rate_limits: crate::middleware::RateLimitConfig,
 }
 
 impl AppState {
@@ -88,7 +90,10 @@ impl IntoResponse for ApiError {
             tracing::warn!(status = %status, error = %error_message, "Client error");
         }
 
-        (status, Json(ErrorResponse { error: error_message, details: None })).into_response()
+        let error_code = self.0.error_code().to_string();
+
+        (status, Json(ErrorResponse { error: error_message, code: error_code, details: None }))
+            .into_response()
     }
 }
 
@@ -203,7 +208,7 @@ pub async fn register(
     if let Some(email_service) = &state.email_service {
         let email_addr = email.email.clone();
         let user_name = payload.name.clone();
-        let token_str = verification_token.token.clone();
+        let token_str = verification_token.secure_token.token.clone();
         let email_service = Arc::clone(email_service);
         let frontend_url = state.config.frontend.url.clone();
 
@@ -784,7 +789,7 @@ mod tests {
         // Get the reset token from the repository
         let tokens = repos.user_password_reset_token.get_by_user(user_id).await.unwrap();
         assert_eq!(tokens.len(), 1);
-        let reset_token = tokens[0].token.clone();
+        let reset_token = tokens[0].secure_token.token.clone();
 
         // Confirm password reset
         let confirm_request = axum::http::Request::builder()
@@ -926,7 +931,7 @@ mod tests {
 
         // Get the reset token
         let tokens = repos.user_password_reset_token.get_by_user(user_id).await.unwrap();
-        let reset_token = tokens[0].token.clone();
+        let reset_token = tokens[0].secure_token.token.clone();
 
         // Confirm password reset
         let confirm_request = axum::http::Request::builder()

crates/api/Cargo.toml

@@ -339,14 +339,15 @@ pub async fn create_certificate(
     let now = Utc::now();
     let public_signing_key = PublicSigningKey {
         kid: cert.kid.clone(),
-        public_key: public_key_base64.clone(),
-        client_id,
-        cert_id: cert.id,
+        public_key: public_key_base64.clone().into(),
+        client_id: client_id.into(),
+        cert_id: cert.id.into(),
         created_at: now,
         valid_from: now,
         valid_until: None,
         active: true,
         revoked_at: None,
+        revocation_reason: None,
     };
 
     // org_id maps directly to namespace_id in Ledger
@@ -361,7 +362,7 @@ pub async fn create_certificate(
 
     // Time the Ledger write operation for metrics
     let ledger_start = std::time::Instant::now();
-    if let Err(e) = signing_key_store.create_key(namespace_id, &public_signing_key).await {
+    if let Err(e) = signing_key_store.create_key(namespace_id.into(), &public_signing_key).await {
         tracing::error!(
             error = %e,
             kid = %cert.kid,
@@ -568,7 +569,7 @@ pub async fn revoke_certificate(
     // Time the Ledger revoke operation for metrics
     let ledger_start = std::time::Instant::now();
     if let Err(e) = signing_key_store
-        .revoke_key(namespace_id, &cert.kid, Some("Certificate revoked by user"))
+        .revoke_key(namespace_id.into(), &cert.kid, Some("Certificate revoked by user"))
         .await
     {
         tracing::error!(
@@ -758,14 +759,15 @@ pub async fn rotate_certificate(
     // Write public key to Ledger with valid_from in the future
     let public_signing_key = PublicSigningKey {
         kid: new_cert.kid.clone(),
-        public_key: public_key_base64.clone(),
-        client_id,
-        cert_id: new_cert.id,
+        public_key: public_key_base64.clone().into(),
+        client_id: client_id.into(),
+        cert_id: new_cert.id.into(),
         created_at: now,
         valid_from,
         valid_until: None,
         active: true,
         revoked_at: None,
+        revocation_reason: None,
     };
 
     // org_id maps directly to namespace_id in Ledger
@@ -781,7 +783,7 @@ pub async fn rotate_certificate(
 
     // Time the Ledger write operation for metrics
     let ledger_start = std::time::Instant::now();
-    if let Err(e) = signing_key_store.create_key(namespace_id, &public_signing_key).await {
+    if let Err(e) = signing_key_store.create_key(namespace_id.into(), &public_signing_key).await {
         tracing::error!(
             error = %e,
             kid = %new_cert.kid,

crates/api/src/handlers/auth.rs

@@ -227,7 +227,7 @@ pub async fn resend_verification(
     // Delete any existing tokens for this email
     let existing_tokens = repos.user_email_verification_token.get_by_email(email_id).await?;
     for token in existing_tokens {
-        repos.user_email_verification_token.delete(token.id).await?;
+        repos.user_email_verification_token.delete(token.secure_token.id).await?;
     }
 
     // Generate new verification token

crates/api/src/handlers/clients.rs

@@ -100,7 +100,7 @@ pub async fn create_team(
         repos.org_team.count_active_by_organization(org_ctx.organization_id).await?;
 
     if current_count >= organization.tier.max_teams() {
-        return Err(CoreError::validation(format!(
+        return Err(CoreError::tier_limit(format!(
             "Team limit reached for tier {:?}. Maximum: {}",
             organization.tier,
             organization.tier.max_teams()

crates/api/src/handlers/emails.rs

@@ -408,36 +408,26 @@ pub async fn client_assertion_authenticate(
     }
 
     // Verify JWT signature using certificate public key
-    use base64::{
-        Engine,
-        engine::general_purpose::{STANDARD as BASE64, URL_SAFE_NO_PAD},
-    };
+    use base64::{Engine, engine::general_purpose::URL_SAFE_NO_PAD};
+    use ed25519_dalek::pkcs8::spki::EncodePublicKey;
     use jsonwebtoken::{Algorithm, DecodingKey, Validation, decode};
 
     let public_key_bytes = URL_SAFE_NO_PAD
         .decode(&certificate.public_key)
         .map_err(|e| CoreError::internal(format!("Failed to decode public key: {e}")))?;
 
-    if public_key_bytes.len() != 32 {
-        return Err(CoreError::internal("Invalid public key length".to_string()).into());
-    }
-
-    // Convert public key to PEM (same as in jwt.rs)
-    let mut spki_der = vec![
-        0x30, 0x2a, // SEQUENCE (42 bytes)
-        0x30, 0x05, // SEQUENCE (algorithm)
-        0x06, 0x03, 0x2b, 0x65, 0x70, // OID 1.3.101.112
-        0x03, 0x21, 0x00, // BIT STRING (33 bytes, 0 unused bits)
-    ];
-    spki_der.extend_from_slice(&public_key_bytes);
-
-    let public_key_pem = format!(
-        "-----BEGIN PUBLIC KEY-----\n{}\n-----END PUBLIC KEY-----\n",
-        BASE64.encode(&spki_der)
-    );
-
-    let decoding_key = DecodingKey::from_ed_pem(public_key_pem.as_bytes())
-        .map_err(|e| CoreError::internal(format!("Failed to create decoding key: {e}")))?;
+    let public_key_array: [u8; 32] = public_key_bytes
+        .as_slice()
+        .try_into()
+        .map_err(|_| CoreError::internal("Invalid public key length".to_string()))?;
+
+    // Encode public key as SPKI DER using the pkcs8/spki crate
+    let verifying_key = ed25519_dalek::VerifyingKey::from_bytes(&public_key_array)
+        .map_err(|e| CoreError::internal(format!("Invalid public key: {e}")))?;
+    let spki_der = verifying_key
+        .to_public_key_der()
+        .map_err(|e| CoreError::internal(format!("Failed to encode SPKI DER: {e}")))?;
+    let decoding_key = DecodingKey::from_ed_der(spki_der.as_ref());
 
     // Set up validation - expect token endpoint as audience
     let mut validation = Validation::new(Algorithm::EdDSA);

crates/api/src/handlers/teams.rs

@@ -140,7 +140,7 @@ pub async fn delete_user(
     for email in &emails {
         let tokens = repos.user_email_verification_token.get_by_email(email.id).await?;
         for token in tokens {
-            repos.user_email_verification_token.delete(token.id).await?;
+            repos.user_email_verification_token.delete(token.secure_token.id).await?;
         }
     }
 
@@ -152,7 +152,7 @@ pub async fn delete_user(
     // CASCADE DELETE: Delete all password reset tokens
     let reset_tokens = repos.user_password_reset_token.get_by_user(ctx.user_id).await?;
     for token in reset_tokens {
-        repos.user_password_reset_token.delete(token.id).await?;
+        repos.user_password_reset_token.delete(token.secure_token.id).await?;
     }
 
     // Soft-delete the user