feat: complete X-Forward-User auth integration for LLM extraction
CI / CI (pull_request) Canceled after 0s

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
This commit is contained in:
2026-09-14 23:49:27 +09:00
parent ff095b4f79
commit db79ea8ffd
3 changed files with 35 additions and 5 deletions
+15 -2
View File
@@ -527,8 +527,19 @@ pub async fn ingest_handler(
INGEST_BYTES_TOTAL.inc_by(byte_count as u64); INGEST_BYTES_TOTAL.inc_by(byte_count as u64);
INGEST_RECORDS_TOTAL.inc_by(body.records.len() 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 // Execute ingest
let resp = execute_ingest(&state, &body).await; let resp = execute_ingest(&state, &body, x_forward_user).await;
INGEST_IN_FLIGHT.dec(); INGEST_IN_FLIGHT.dec();
resp resp
} }
@@ -537,6 +548,7 @@ pub async fn ingest_handler(
async fn execute_ingest( async fn execute_ingest(
state: &web::Data<AppState>, state: &web::Data<AppState>,
body: &IngestRequest, body: &IngestRequest,
x_forward_user: Option<String>,
) -> HttpResponse { ) -> HttpResponse {
let records: Vec<(String, String)> = body.records let records: Vec<(String, String)> = body.records
.iter() .iter()
@@ -567,8 +579,9 @@ async fn execute_ingest(
let worker = state.ingest_worker.clone(); let worker = state.ingest_worker.clone();
let project = body.project.clone(); let project = body.project.clone();
let ingest_id = body.ingest_id.clone(); let ingest_id = body.ingest_id.clone();
let x_fwd = x_forward_user.clone();
tokio::spawn(async move { 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); tracing::error!("Ingest failed: {}", e);
} }
}); });
+13 -1
View File
@@ -67,6 +67,17 @@ impl IngestWorker {
project: &str, project: &str,
ingest_id: &str, ingest_id: &str,
records: Vec<(String, String)>, // (content, source) 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<String>,
) -> Result<()> { ) -> Result<()> {
tracing::info!( tracing::info!(
target: "ingest", target: "ingest",
@@ -119,7 +130,8 @@ impl IngestWorker {
}; };
// Run extraction pipeline (entity + fact extraction + contradiction detection) // 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) => { Ok(result) => {
tracing::debug!( tracing::debug!(
target: "ingest", target: "ingest",
+7 -2
View File
@@ -59,10 +59,15 @@ impl IngestPipeline {
/// Execute extraction pipeline for episode /// Execute extraction pipeline for episode
/// CRAP: 14 (Low: orchestration only, delegates to stages) /// CRAP: 14 (Low: orchestration only, delegates to stages)
pub async fn ingest(&self, episode: &Episode) -> Result<ExtractionResult> { pub async fn ingest(&self, episode: &Episode) -> Result<ExtractionResult> {
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<ExtractionResult> {
debug!("Starting ingest for episode: {}", episode.id); debug!("Starting ingest for episode: {}", episode.id);
// Stage 1: Extract entities // Stage 1: Extract entities (with optional auth header)
let extracted_entities = self.entity_extractor.extract(&episode.text).await?; let extracted_entities = self.entity_extractor.extract_with_auth(&episode.text, x_forward_user).await?;
debug!("Extracted {} entities", extracted_entities.len()); debug!("Extracted {} entities", extracted_entities.len());
// Convert to domain entities // Convert to domain entities