RustProxy/src/admin/file_logs.rs

const FILE_LOG_SCAN_BYTES_PER_SOURCE: u64 = 2 * 1024 * 1024;
const FILE_LOG_SCAN_BYTES_TOTAL: u64 = 8 * 1024 * 1024;
const FILE_LOG_SCAN_LINES: usize = 20_000;
const FILE_LOG_MAX_SOURCES: usize = 32;
const FILE_LOG_MAX_LIMIT: usize = 500;
const FILE_LOG_MAX_OFFSET: usize = 10_000;
const FILE_LOG_GZIP_INPUT_MAX: u64 = 8 * 1024 * 1024;
const FILE_LOG_GZIP_OUTPUT_MAX: u64 = 64 * 1024 * 1024;
#[derive(Clone)]
struct FileLogSource {
    id: String,
    label: String,
    kind: &'static str,
    path: std::path::PathBuf,
    enabled: bool,
    format: &'static str,
    exists: bool,
    size: u64,
    modified_ms: u64,
    compressed: bool,
}
#[derive(Default, Deserialize)]
#[serde(default, deny_unknown_fields)]
struct FileLogQuery {
    source: Option<String>,
    q: Option<String>,
    event: Option<String>,
    status_class: Option<String>,
    client_ip: Option<String>,
    vhost: Option<String>,
    offset: Option<usize>,
    limit: Option<usize>,
}
fn file_modified_ms(metadata: &std::fs::Metadata) -> u64 {
    metadata
        .modified()
        .ok()
        .and_then(|time| time.duration_since(std::time::UNIX_EPOCH).ok())
        .map(|duration| duration.as_millis() as u64)
        .unwrap_or(0)
}
fn source_from_path(
    id: String,
    label: String,
    kind: &'static str,
    path: std::path::PathBuf,
    enabled: bool,
    format: &'static str,
) -> FileLogSource {
    let metadata = std::fs::symlink_metadata(&path).ok();
    let safe_regular = metadata
        .as_ref()
        .is_some_and(|meta| meta.file_type().is_file() && !meta.file_type().is_symlink());
    let compressed = path
        .extension()
        .and_then(|ext| ext.to_str())
        .is_some_and(|ext| ext.eq_ignore_ascii_case("gz"));
    FileLogSource {
        id,
        label,
        kind,
        path,
        enabled,
        format,
        exists: safe_regular,
        size: metadata
            .as_ref()
            .filter(|_| safe_regular)
            .map_or(0, |m| m.len()),
        modified_ms: metadata
            .as_ref()
            .filter(|_| safe_regular)
            .map_or(0, file_modified_ms),
        compressed,
    }
}
include!("file_logs_io.rs");

fn configured_file_log_sources(config: &crate::config::Config) -> Vec<FileLogSource> {
    let mut sources = Vec::new();
    if config.logging.file.per_vhost {
        let directory = std::path::PathBuf::from(&config.logging.file.directory);
        let directory_safe = std::fs::symlink_metadata(&directory)
            .ok()
            .is_some_and(|meta| meta.file_type().is_dir() && !meta.file_type().is_symlink());
        if directory_safe {
            let entries = newest_log_files(std::fs::read_dir(&directory)
                .ok()
                .into_iter()
                .flat_map(|iter| iter.filter_map(|entry| entry.ok()))
                .filter_map(|entry| {
                    let path = entry.path();
                    let name = entry.file_name().to_string_lossy().into_owned();
                    let plain = name.ends_with(".jsonl");
                    let rotated = name.contains(".jsonl.") && name.ends_with(".gz");
                    if (!plain && !rotated) || name.len() > 180 {
                        return None;
                    }
                    let file_type = entry.file_type().ok()?;
                    if !file_type.is_file() || file_type.is_symlink() {
                        return None;
                    }
                    let meta = entry.metadata().ok()?;
                    Some((name, path, file_modified_ms(&meta)))
                }), 256);
            for (name, path, _) in entries {
                let id = if name.ends_with(".jsonl") {
                    format!("vhost:{}", name.trim_end_matches(".jsonl"))
                } else {
                    format!("vhost-gz:{name}")
                };
                sources.push(source_from_path(
                    id,
                    name,
                    "access",
                    path,
                    config.logging.file.enabled,
                    "JSONL",
                ));
            }
        }
    } else {
        let access_path = std::path::PathBuf::from(&config.logging.file.path);
        sources.push(source_from_path(
            "access".into(),
            "Access log".into(),
            "access",
            access_path.clone(),
            config.logging.file.enabled,
            "JSONL",
        ));
        sources.extend(rotated_gzip_sources(&access_path, "access", "Access log", "access", config.logging.file.enabled, "JSONL"));
    }
    let error_format = match config.error_log.format {
        crate::logging::ErrorLogFormat::Json => "JSONL",
        crate::logging::ErrorLogFormat::Raw => "RAW",
    };
    let error_path = std::path::PathBuf::from(&config.error_log.path);
    sources.push(source_from_path(
        "error".into(),
        "Error log".into(),
        "error",
        error_path.clone(),
        config.error_log.enabled,
        error_format,
    ));
    sources.extend(rotated_gzip_sources(&error_path, "error", "Error log", "error", config.error_log.enabled, error_format));
    sources
}
fn file_log_sources_json(sources: &[FileLogSource]) -> Value {
    Value::Array(
        sources
            .iter()
            .map(|source| {
                json!({
                    "id": source.id,
                    "label": source.label,
                    "kind": source.kind,
                    "path": source.path.to_string_lossy(),
                    "enabled": source.enabled,
                    "format": source.format,
                    "exists": source.exists,
                    "size": source.size,
                    "modified_ms": source.modified_ms,
                    "compressed": source.compressed,
                })
            })
            .collect(),
    )
}
async fn file_logs(State(state): State<Arc<App>>) -> Result<Json<Value>, ApiError> {
    let config = state.config.load().envelope.config.clone();
    let sources = tokio::task::spawn_blocking(move || configured_file_log_sources(&config))
        .await
        .map_err(|_| {
            ApiError(
                StatusCode::INTERNAL_SERVER_ERROR,
                "file log inventory worker failed".into(),
            )
        })?;
    Ok(Json(json!({
        "sources": file_log_sources_json(&sources),
        "limits": {
            "max_limit": FILE_LOG_MAX_LIMIT,
            "max_offset": FILE_LOG_MAX_OFFSET,
            "scan_bytes_per_source": FILE_LOG_SCAN_BYTES_PER_SOURCE,
            "scan_bytes_total": FILE_LOG_SCAN_BYTES_TOTAL,
            "scan_lines": FILE_LOG_SCAN_LINES,
            "max_sources_per_query": FILE_LOG_MAX_SOURCES,
        }
    })))
}
fn copy_mapped_field(
    object: &mut serde_json::Map<String, Value>,
    format: &crate::logging::JsonFormat,
    canonical: &str,
) {
    if object.contains_key(canonical) {
        return;
    }
    let mapped = format
        .fields
        .iter()
        .find_map(|(output, source)| (source == canonical).then_some(output));
    if let Some(value) = mapped.and_then(|output| object.get(output)).cloned() {
        object.insert(canonical.to_owned(), value);
    }
}

fn normalized_record(
    source: &FileLogSource,
    line: &str,
    format: &crate::logging::JsonFormat,
) -> Value {
    let mut value = if source.format == "RAW" {
        parse_raw_error_record(line)
    } else {
        serde_json::from_str::<Value>(line).unwrap_or_else(|_| json!({"raw": line}))
    };
    if !value.is_object() {
        value = json!({"value": value});
    }
    let object = value.as_object_mut().expect("record normalized to object");
    if source.kind == "access" {
        for field in [
            "event",
            "time_ms",
            "status",
            "client_ip",
            "vhost",
            "method",
            "host",
            "path",
            "transport",
            "http_version",
            "protocol_error",
            "error_detail",
            "tls_sni",
            "reason",
            "error",
            "sni",
            "authority",
            "matched_by",
        ] {
            copy_mapped_field(object, format, field);
        }
        if !object.contains_key("client_ip") {
            if let Some(value) = object.get("source_ip").cloned() {
                object.insert("client_ip".into(), value);
            }
        }
        if !object.contains_key("reason") {
            if let Some(value) = object.get("protocol_error").cloned() {
                object.insert("reason".into(), value);
            }
        }
        if !object.contains_key("error") {
            if let Some(value) = object.get("error_detail").cloned() {
                object.insert("error".into(), value);
            }
        }
        if !object.contains_key("sni") {
            if let Some(value) = object.get("tls_sni").cloned() {
                object.insert("sni".into(), value);
            }
        }
    }
    object.entry("event").or_insert_with(|| {
        Value::String(
            if source.kind == "error" {
                "error"
            } else {
                "request"
            }
            .into(),
        )
    });
    object.insert("_source".into(), Value::String(source.id.clone()));
    object.insert("_source_kind".into(), Value::String(source.kind.into()));
    object.insert("_source_format".into(), Value::String(source.format.into()));
    value
}

fn contains_ci(value: Option<&Value>, needle: &str) -> bool {
    value
        .and_then(Value::as_str)
        .is_some_and(|text| text.to_ascii_lowercase().contains(needle))
}

fn record_matches(value: &Value, query: &FileLogQuery, search: &[FileLogSearchClause]) -> bool {
    let Some(object) = value.as_object() else {
        return false;
    };
    if !record_matches_search(value, search) {
        return false;
    }
    if let Some(event) = query
        .event
        .as_deref()
        .filter(|v| !v.is_empty() && *v != "all")
    {
        if object.get("event").and_then(Value::as_str) != Some(event) {
            return false;
        }
    }
    if let Some(class) = query
        .status_class
        .as_deref()
        .filter(|v| !v.is_empty() && *v != "all")
    {
        let Some(status) = object.get("status").and_then(Value::as_u64) else {
            return false;
        };
        let expected = class
            .as_bytes()
            .first()
            .and_then(|v| v.checked_sub(b'0'))
            .map(u64::from);
        if expected != Some(status / 100) {
            return false;
        }
    }
    if let Some(client) = query.client_ip.as_deref().filter(|v| !v.is_empty()) {
        if !contains_ci(object.get("client_ip"), &client.to_ascii_lowercase()) {
            return false;
        }
    }
    if let Some(vhost) = query.vhost.as_deref().filter(|v| !v.is_empty()) {
        if !contains_ci(object.get("vhost"), &vhost.to_ascii_lowercase()) {
            return false;
        }
    }
    true
}

fn validate_file_log_query(query: &FileLogQuery) -> Result<(), ApiError> {
    let text_ok = |value: &Option<String>, max: usize| {
        value
            .as_ref()
            .is_none_or(|v| v.len() <= max && !v.chars().any(char::is_control))
    };
    if !text_ok(&query.source, 192)
        || !text_ok(&query.q, 512)
        || !text_ok(&query.client_ip, 128)
        || !text_ok(&query.vhost, 256)
        || query.offset.unwrap_or(0) > FILE_LOG_MAX_OFFSET
        || query
            .limit
            .is_some_and(|v| v == 0 || v > FILE_LOG_MAX_LIMIT)
        || query.event.as_ref().is_some_and(|v| {
            !matches!(
                v.as_str(),
                "" | "all" | "request" | "protocol_error" | "error"
            )
        })
        || query.status_class.as_ref().is_some_and(|v| {
            !matches!(
                v.as_str(),
                "" | "all" | "1xx" | "2xx" | "3xx" | "4xx" | "5xx"
            )
        })
    {
        return Err(ApiError(
            StatusCode::BAD_REQUEST,
            "invalid file log filter".into(),
        ));
    }
    Ok(())
}

async fn file_logs_query(
    State(state): State<Arc<App>>,
    Query(query): Query<FileLogQuery>,
) -> Result<Json<Value>, ApiError> {
    validate_file_log_query(&query)?;
    let search = parse_file_log_search(query.q.as_deref().unwrap_or(""))?;
    let config = state.config.load().envelope.config.clone();
    tokio::task::spawn_blocking(move || {
        let mut sources = configured_file_log_sources(&config);
        let log_format = config.logging.format.clone();
        let requested = query.source.as_deref().unwrap_or("all");
        let mut truncated = false;
        if requested != "all" {
            sources.retain(|source| source.id == requested);
            if sources.is_empty() {
                return Err(ApiError(
                    StatusCode::BAD_REQUEST,
                    "unknown file log source".into(),
                ));
            }
        } else {
            sources.sort_by(|a, b| b.modified_ms.cmp(&a.modified_ms));
            if sources.len() > FILE_LOG_MAX_SOURCES {
                truncated = true;
                sources.truncate(FILE_LOG_MAX_SOURCES);
            }
        }
        let offset = query.offset.unwrap_or(0);
        let limit = query.limit.unwrap_or(100).clamp(1, FILE_LOG_MAX_LIMIT);
        let mut remaining_bytes = FILE_LOG_SCAN_BYTES_TOTAL;
        let mut scanned_bytes = 0u64;
        let mut remaining_lines = FILE_LOG_SCAN_LINES;
        let mut scanned_lines = 0usize;
        let mut records = Vec::new();
        let mut fields = std::collections::BTreeSet::new();
        for (index, source) in sources.iter().enumerate() {
            if remaining_bytes == 0 || remaining_lines == 0 {
                truncated = true;
                break;
            }
            let sources_left = sources.len().saturating_sub(index).max(1);
            let line_budget = (remaining_lines / sources_left).max(1);
            match read_tail(source, remaining_bytes, line_budget) {
                Ok((lines, bytes, source_truncated)) => {
                    remaining_bytes = remaining_bytes.saturating_sub(bytes);
                    remaining_lines = remaining_lines.saturating_sub(lines.len());
                    scanned_bytes += bytes;
                    scanned_lines += lines.len();
                    truncated |= source_truncated;
                    for line in lines {
                        if line.is_empty() {
                            continue;
                        }
                        let value = normalized_record(source, &line, &log_format);
                        if record_matches(&value, &query, &search) {
                            collect_file_log_fields(&value, &mut fields);
                            records.push(value);
                        }
                    }
                }
                Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
                Err(_) => truncated = true,
            }
        }
        records.sort_by(|a, b| {
            let numeric = |value: &Value| value.get("time_ms").and_then(Value::as_u64);
            match (numeric(a), numeric(b)) {
                (Some(left), Some(right)) => right.cmp(&left),
                _ => b
                    .get("timestamp")
                    .and_then(Value::as_str)
                    .unwrap_or("")
                    .cmp(a.get("timestamp").and_then(Value::as_str).unwrap_or("")),
            }
        });
        let total = records.len();
        let page = records
            .into_iter()
            .skip(offset)
            .take(limit)
            .collect::<Vec<_>>();
        Ok(Json(json!({
            "entries": page,
            "total": total,
            "offset": offset,
            "limit": limit,
            "truncated": truncated,
            "scanned_bytes": scanned_bytes,
            "scanned_lines": scanned_lines,
            "fields": fields,
            "sources": file_log_sources_json(&sources),
        })))
    })
    .await
    .map_err(|_| {
        ApiError(
            StatusCode::INTERNAL_SERVER_ERROR,
            "file log query worker failed".into(),
        )
    })?
}