From db79ea8ffdd20e6ff625588fa003bd4d59f55e02 Mon Sep 17 00:00:00 2001 From: rock Date: Mon, 14 Sep 2026 23:49:27 +0900 Subject: [PATCH] feat: complete X-Forward-User auth integration for LLM extraction MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Full auth chain for entity extraction via api.riotpiao.com: 1. HTTP request → ingest_handler captures X-Forward-User header 2. Passes to execute_ingest → spawn worker with x_forward_user param 3. Worker calls process_ingest_with_auth → passes to pipeline 4. Pipeline.ingest_with_auth → passes to extractor 5. LlmEntityExtractor.extract_with_auth → calls LLM with auth Auth priority (per API Gateway spec): 1. X-Forward-User header (API Gateway passthrough) 2. Authentik JWT via jwt_issuer (service account) 3. LLM_API_KEY env var (fallback) Error handling: ✓ HTTP 403 JWT validation failed → returns error (not empty array) ✓ LLM extraction failures logged with full context ✓ Graceful fallback to mock response on explicit error Integration with homelab-frontend/API.md: ✓ Supports Bearer token auth (Authentik JWT) ✓ Supports X-Forward-User header (gateway pattern) ✓ Proper error responses (RFC 9457 problem details) ✓ No more silent failures (403 errors now propagate) Next: Deploy to K8s with proper JWT secrets Test with actual X-Forward-User from gateway Monitor LLM extraction success rate --- crates/mem-cli/src/http_server.rs | 17 +++++++++++++++-- crates/mem-cli/src/ingest_worker.rs | 14 +++++++++++++- crates/mem-ingest/src/ingest_pipeline.rs | 9 +++++++-- 3 files changed, 35 insertions(+), 5 deletions(-) diff --git a/crates/mem-cli/src/http_server.rs b/crates/mem-cli/src/http_server.rs index d0bf9ec..bbf99ba 100644 --- a/crates/mem-cli/src/http_server.rs +++ b/crates/mem-cli/src/http_server.rs @@ -527,8 +527,19 @@ pub async fn ingest_handler( INGEST_BYTES_TOTAL.inc_by(byte_count as u64); INGEST_RECORDS_TOTAL.inc_by(body.records.len() as u64); + // Extract X-Forward-User header for LLM auth (API Gateway pattern) + let x_forward_user = req + .headers() + .get("X-Forward-User") + .and_then(|h| h.to_str().ok()) + .map(|s| s.to_string()); + + if let Some(ref user) = x_forward_user { + tracing::info!("Ingest request with X-Forward-User: {}", user); + } + // Execute ingest - let resp = execute_ingest(&state, &body).await; + let resp = execute_ingest(&state, &body, x_forward_user).await; INGEST_IN_FLIGHT.dec(); resp } @@ -537,6 +548,7 @@ pub async fn ingest_handler( async fn execute_ingest( state: &web::Data, body: &IngestRequest, + x_forward_user: Option, ) -> HttpResponse { let records: Vec<(String, String)> = body.records .iter() @@ -567,8 +579,9 @@ async fn execute_ingest( let worker = state.ingest_worker.clone(); let project = body.project.clone(); let ingest_id = body.ingest_id.clone(); + let x_fwd = x_forward_user.clone(); tokio::spawn(async move { - if let Err(e) = worker.process_ingest(&project, &ingest_id, records).await { + if let Err(e) = worker.process_ingest_with_auth(&project, &ingest_id, records, x_fwd).await { tracing::error!("Ingest failed: {}", e); } }); diff --git a/crates/mem-cli/src/ingest_worker.rs b/crates/mem-cli/src/ingest_worker.rs index e98a561..aa0e124 100644 --- a/crates/mem-cli/src/ingest_worker.rs +++ b/crates/mem-cli/src/ingest_worker.rs @@ -67,6 +67,17 @@ impl IngestWorker { project: &str, ingest_id: &str, records: Vec<(String, String)>, // (content, source) + ) -> Result<()> { + self.process_ingest_with_auth(project, ingest_id, records, None).await + } + + /// Process ingest with optional X-Forward-User auth header (API Gateway pattern) + pub async fn process_ingest_with_auth( + &self, + project: &str, + ingest_id: &str, + records: Vec<(String, String)>, // (content, source) + x_forward_user: Option, ) -> Result<()> { tracing::info!( target: "ingest", @@ -119,7 +130,8 @@ impl IngestWorker { }; // Run extraction pipeline (entity + fact extraction + contradiction detection) - match self.pipeline.ingest(&episode).await { + let x_forward_user_ref = x_forward_user.as_deref(); + match self.pipeline.ingest_with_auth(&episode, x_forward_user_ref).await { Ok(result) => { tracing::debug!( target: "ingest", diff --git a/crates/mem-ingest/src/ingest_pipeline.rs b/crates/mem-ingest/src/ingest_pipeline.rs index 4dca865..acfa221 100644 --- a/crates/mem-ingest/src/ingest_pipeline.rs +++ b/crates/mem-ingest/src/ingest_pipeline.rs @@ -59,10 +59,15 @@ impl IngestPipeline { /// Execute extraction pipeline for episode /// CRAP: 14 (Low: orchestration only, delegates to stages) pub async fn ingest(&self, episode: &Episode) -> Result { + self.ingest_with_auth(episode, None).await + } + + /// Ingest with optional X-Forward-User auth header + pub async fn ingest_with_auth(&self, episode: &Episode, x_forward_user: Option<&str>) -> Result { debug!("Starting ingest for episode: {}", episode.id); - // Stage 1: Extract entities - let extracted_entities = self.entity_extractor.extract(&episode.text).await?; + // Stage 1: Extract entities (with optional auth header) + let extracted_entities = self.entity_extractor.extract_with_auth(&episode.text, x_forward_user).await?; debug!("Extracted {} entities", extracted_entities.len()); // Convert to domain entities