RustProxy/src/admin/logging_api.rs

async fn update_section(
    State(state): State<Arc<App>>,
    Extension(session): Extension<Session>,
    Path(section): Path<String>,
    Json(input): Json<SectionUpdate>,
) -> Result<Json<crate::config::Envelope>, ApiError> {
    let current = state.config.load_full();
    if current.envelope.revision != input.expected_revision {
        return Err(ApiError(
            StatusCode::CONFLICT,
            "configuration changed; reload before saving".into(),
        ));
    }
    let mut config = current.envelope.config.clone();
    let value = input.value;
    match section.as_str() {
        "listeners" => {
            config.listeners =
                serde_json::from_value(value).map_err(|e| invalid_section(&section, e))?
        }
        "docker-agents" => {
            config.docker_agents =
                serde_json::from_value(value).map_err(|e| invalid_section(&section, e))?
        }
        "http2" => {
            config.http2 =
                serde_json::from_value(value).map_err(|e| invalid_section(&section, e))?
        }
        "response" => {
            let part: ResponseSection =
                serde_json::from_value(value).map_err(|e| invalid_section(&section, e))?;
            config.server_header = part.server_header;
            config.diagnostic_headers = part.diagnostic_headers;
            config.compression = part.compression;
        }
        "pages" => {
            config.pages =
                serde_json::from_value(value).map_err(|e| invalid_section(&section, e))?
        }
        "cache" => {
            config.cache =
                serde_json::from_value(value).map_err(|e| invalid_section(&section, e))?
        }
        "transform" => {
            config.transform =
                serde_json::from_value(value).map_err(|e| invalid_section(&section, e))?
        }
        "logging" => {
            if value.get("logging").is_some() || value.get("error_log").is_some() {
                let part: LoggingSection =
                    serde_json::from_value(value).map_err(|e| invalid_section(&section, e))?;
                config.logging = part.logging;
                config.error_log = part.error_log;
            } else {
                // Backward compatibility for clients that still send only the
                // historical access-logging object to /api/config/logging.
                config.logging =
                    serde_json::from_value(value).map_err(|e| invalid_section(&section, e))?;
            }
        }
        "forwarding" => {
            config.forwarding =
                serde_json::from_value(value).map_err(|e| invalid_section(&section, e))?
        }
        "headers" => {
            config.headers =
                serde_json::from_value(value).map_err(|e| invalid_section(&section, e))?
        }
        "prometheus" => {
            config.prometheus =
                serde_json::from_value(value).map_err(|e| invalid_section(&section, e))?
        }
        "administration" => {
            config.administration =
                serde_json::from_value(value).map_err(|e| invalid_section(&section, e))?
        }
        "tls" => {
            config.tls = serde_json::from_value(value).map_err(|e| invalid_section(&section, e))?
        }
        "vhosts" => {
            config.vhosts =
                serde_json::from_value(value).map_err(|e| invalid_section(&section, e))?
        }
        "web-services" => {
            config.web_services =
                serde_json::from_value(value).map_err(|e| invalid_section(&section, e))?
        }
        "redirects" => {
            config.redirects =
                serde_json::from_value(value).map_err(|e| invalid_section(&section, e))?
        }
        "not_found_hosts" => {
            config.not_found_hosts =
                serde_json::from_value(value).map_err(|e| invalid_section(&section, e))?
        }
        "streams" => {
            config.streams =
                serde_json::from_value(value).map_err(|e| invalid_section(&section, e))?
        }
        "acl-profiles" => {
            config.acl_profiles =
                serde_json::from_value(value).map_err(|e| invalid_section(&section, e))?
        }
        "host-policies" => {
            config.host_policies =
                serde_json::from_value(value).map_err(|e| invalid_section(&section, e))?
        }
        _ => {
            return Err(ApiError(
                StatusCode::NOT_FOUND,
                "unknown configuration section".into(),
            ))
        }
    }
    state
        .update_as(
            Update {
                expected_revision: input.expected_revision,
                config,
            },
            session.username,
        )
        .await
        .map(Json)
}
fn logs_payload(state: &App) -> Value {
    json!({
        "entries": state.logs.recent_limit(crate::logging::LIVE_WINDOW),
        "capacity": crate::logging::HISTORY_CAPACITY,
        "window": crate::logging::LIVE_WINDOW,
        "dropped": state.logs.dropped.load(Ordering::Relaxed)
    })
}
#[derive(Default, Deserialize)]
#[serde(default, deny_unknown_fields)]
struct LogFilter {
    vhost: Option<String>,
    q: Option<String>,
    status: Option<u16>,
    cache: Option<String>,
    limit: Option<usize>,
}
async fn logs(
    State(state): State<Arc<App>>,
    Query(filter): Query<LogFilter>,
) -> Result<Json<Value>, ApiError> {
    if filter.q.as_ref().is_some_and(|q| q.len() > 256)
        || filter
            .vhost
            .as_ref()
            .is_some_and(|v| !crate::config::valid_id(v))
        || filter
            .cache
            .as_ref()
            .is_some_and(|v| !matches!(v.as_str(), "HIT" | "MISS" | "BYPASS" | "-"))
    {
        return Err(ApiError(
            StatusCode::BAD_REQUEST,
            "invalid log filter".into(),
        ));
    }
    let q = filter.q.unwrap_or_default().to_ascii_lowercase();
    let limit = filter
        .limit
        .unwrap_or(crate::logging::LIVE_WINDOW)
        .clamp(1, crate::logging::HISTORY_CAPACITY);
    let entries: Vec<_> = state
        .logs
        .recent()
        .into_iter()
        .filter(|e| {
            filter
                .vhost
                .as_ref()
                .is_none_or(|v| e.vhost.as_ref() == Some(v))
                && filter.status.is_none_or(|v| e.status == v)
                && filter.cache.as_ref().is_none_or(|v| e.cache == v.as_str())
                && (q.is_empty()
                    || format!(
                        "{} {} {} {}",
                        e.method,
                        e.host,
                        e.path,
                        e.backend.as_deref().unwrap_or("")
                    )
                    .to_ascii_lowercase()
                    .contains(&q))
        })
        .take(limit)
        .collect();
    Ok(Json(json!({
        "entries": entries,
        "capacity": crate::logging::HISTORY_CAPACITY,
        "window": crate::logging::LIVE_WINDOW,
        "dropped": state.logs.dropped.load(Ordering::Relaxed)
    })))
}

async fn live_logs_stream(
    State(state): State<Arc<App>>,
    headers: HeaderMap,
) -> Result<Response, ApiError> {
    let secret = auth::cookie_secret(&headers)
        .ok_or_else(|| ApiError(StatusCode::UNAUTHORIZED, "auth.required".into()))?;
    let permit = state
        .viewer_slots
        .clone()
        .try_acquire_owned()
        .map_err(|_| {
            ApiError(
                StatusCode::TOO_MANY_REQUESTS,
                "events.too_many_viewers".into(),
            )
        })?;
    let receiver = state.logs.subscribe();
    let next_auth_check = tokio::time::Instant::now() + Duration::from_secs(10);
    let stream = futures_util::stream::unfold(
        (state, secret, permit, receiver, next_auth_check),
        |(state, secret, permit, mut receiver, mut next_auth_check)| async move {
            if tokio::time::Instant::now() >= next_auth_check {
                let session_seconds = state
                    .config
                    .load()
                    .envelope
                    .config
                    .administration
                    .session_seconds();
                match state
                    .store
                    .session_at_with_browser_lifetime(&secret, auth::now(), session_seconds)
                    .await
                {
                    Ok(Some(_)) => {
                        next_auth_check = tokio::time::Instant::now() + Duration::from_secs(10);
                    }
                    _ => return None,
                }
            }
            let event = tokio::select! {
                received = receiver.recv() => match received {
                    Ok(entry) => Event::default()
                        .event("log")
                        .data(serde_json::to_string(&entry).unwrap_or_else(|_| "{}".into())),
                    Err(tokio::sync::broadcast::error::RecvError::Lagged(dropped)) => {
                        Event::default().event("gap").data(json!({"dropped": dropped}).to_string())
                    }
                    Err(tokio::sync::broadcast::error::RecvError::Closed) => return None,
                },
                _ = tokio::time::sleep_until(next_auth_check) => {
                    let session_seconds = state
                        .config
                        .load()
                        .envelope
                        .config
                        .administration
                        .session_seconds();
                    match state
                        .store
                        .session_at_with_browser_lifetime(&secret, auth::now(), session_seconds)
                        .await
                    {
                        Ok(Some(_)) => {
                            next_auth_check = tokio::time::Instant::now() + Duration::from_secs(10);
                            Event::default().event("heartbeat").data("{}")
                        }
                        _ => return None,
                    }
                }
            };
            Some((
                Ok::<Event, Infallible>(event),
                (state, secret, permit, receiver, next_auth_check),
            ))
        },
    );
    Ok(Sse::new(stream)
        .keep_alive(KeepAlive::new().interval(Duration::from_secs(5)))
        .into_response())
}

async fn audit(State(state): State<Arc<App>>) -> Result<Json<Value>, ApiError> {
    Ok(Json(
        json!({"entries":state.store.recent_audit().await.map_err(auth::internal)?,"capacity":crate::db::AUDIT_CAPACITY}),
    ))
}
async fn metrics(State(state): State<Arc<App>>) -> Response {
    crate::prometheus::response(&state)
}
async fn events(State(state): State<Arc<App>>, headers: HeaderMap) -> Result<Response, ApiError> {
    let secret = auth::cookie_secret(&headers)
        .ok_or_else(|| ApiError(StatusCode::UNAUTHORIZED, "auth.required".into()))?;
    let permit = state
        .viewer_slots
        .clone()
        .try_acquire_owned()
        .map_err(|_| {
            ApiError(
                StatusCode::TOO_MANY_REQUESTS,
                "events.too_many_viewers".into(),
            )
        })?;
    // One bounded snapshot, produced only when the consumer polls. No per-client queue.
    // Absolute expiry/revocation is rechecked on every snapshot; no immortal SSE sessions.
    let stream = futures_util::stream::unfold(
        (state, secret, permit, false),
        |(state, secret, permit, delayed)| async move {
            if delayed {
                tokio::time::sleep(Duration::from_secs(2)).await;
            }
            let session_seconds = state
                .config
                .load()
                .envelope
                .config
                .administration
                .session_seconds();
            match state
                .store
                .session_at_with_browser_lifetime(&secret, auth::now(), session_seconds)
                .await
            {
                Ok(Some(_)) => {}
                _ => return None,
            }
            let payload =
                json!({"status":status_payload(&state), "logs":logs_payload(&state)}).to_string();
            let event = Event::default().event("snapshot").data(payload);
            Some((
                Ok::<Event, Infallible>(event),
                (state, secret, permit, true),
            ))
        },
    );
    Ok(Sse::new(stream)
        .keep_alive(KeepAlive::new().interval(Duration::from_secs(10)))
        .into_response())
}

#[derive(Default, Deserialize)]
#[serde(default, deny_unknown_fields)]
struct TrafficQuery {
    vhost: Option<String>,
}
async fn traffic_status(
    State(state): State<Arc<App>>,
    Query(query): Query<TrafficQuery>,
) -> Result<Json<Value>, ApiError> {
    let snapshot = state.config.load_full();
    if query.vhost.as_ref().is_some_and(|id| {
        id != "*"
            && !snapshot
                .envelope
                .config
                .vhosts
                .iter()
                .any(|vhost| &vhost.id == id)
    }) {
        return Err(ApiError(
            StatusCode::NOT_FOUND,
            "unknown vhost traffic scope".into(),
        ));
    }
    let scope = query.vhost.as_deref().unwrap_or("*");
    let vhosts = snapshot
        .envelope
        .config
        .vhosts
        .iter()
        .map(|vhost| json!({"id": vhost.id, "hosts": vhost.frontend.hosts}))
        .collect::<Vec<_>>();
    Ok(Json(json!({
        "scope": scope,
        "since": "process_start",
        "settings": {
            "capture_user_agent": snapshot.envelope.config.logging.capture_user_agent,
            "capture_country": snapshot.envelope.config.logging.capture_country
        },
        "geoip": state.geoip.view(),
        "analytics": state.traffic.view(Some(scope)),
        "vhosts": vhosts
    })))
}

#[derive(Default, Deserialize)]
#[serde(default, deny_unknown_fields)]
struct CacheQuery {
    vhost: Option<String>,
    offset: Option<usize>,
    limit: Option<usize>,
}
async fn cache_status(
    State(state): State<Arc<App>>,
    Query(query): Query<CacheQuery>,
) -> Result<Json<Value>, ApiError> {
    let snapshot = state.config.load_full();
    if query
        .vhost
        .as_ref()
        .is_some_and(|id| !snapshot.envelope.config.vhosts.iter().any(|v| &v.id == id))
    {
        return Err(ApiError(
            StatusCode::NOT_FOUND,
            "unknown vhost cache scope".into(),
        ));
    }
    let scopes: Vec<_> = snapshot.envelope.config.vhosts.iter().map(|vhost| {
        json!({"id":vhost.id, "hosts":vhost.frontend.hosts, "metrics":state.cache.scope_view(&vhost.id),
            "settings":vhost.cache.resolve(&snapshot.envelope.config.cache).ok()})
    }).collect();
    let dashboard = state.cache.dashboard(
        query.vhost.as_deref(),
        query.offset.unwrap_or(0).min(16384),
        query.limit.unwrap_or(50),
    );
    Ok(Json(
        json!({"settings":snapshot.envelope.config.cache, "metrics":state.cache.view(), "vhosts":scopes, "dashboard":dashboard, "bans":state.cache.bans()}),
    ))
}