Source code for aidputils.agents.tools.http.audit_logger

"""
Audit Logger for HTTP Tool.

Provides structured audit logging with:
- Request ID tracking (opc_request_id)
- Timestamp logging (start, end, elapsed_ms)
- Request/response metadata logging
- Credential masking for sensitive headers
"""

import logging
import re
import time
import uuid
from typing import Any, Dict, Optional

logger = logging.getLogger(__name__)

# Headers that contain sensitive credentials and should be masked
SENSITIVE_HEADERS = {
    "authorization",
    "x-api-key",
    "x-auth-token",
    "cookie",
    "set-cookie",
    "proxy-authorization",
}

# Pattern for headers that start with sensitive prefixes
SENSITIVE_HEADER_PREFIXES = (
    "x-auth-",
    "x-api-",
)

# Masking value with clear indication
REDACTED_VALUE = "***[REDACTED]***"




def generate_request_id() -> str:
    """
    Generate a unique request ID for tracking.
    
    Returns:
        UUID string for request tracking
    """
    return str(uuid.uuid4())





def is_sensitive_header(header_name: str) -> bool:
    """
    Check if a header name is sensitive and should be masked.
    
    Args:
        header_name: The header name to check
        
    Returns:
        True if the header is sensitive
    """
    header_lower = header_name.lower()
    
    if header_lower in SENSITIVE_HEADERS:
        return True
    
    for prefix in SENSITIVE_HEADER_PREFIXES:
        if header_lower.startswith(prefix):
            return True
    
    return False





def mask_headers(headers: Dict[str, str]) -> Dict[str, str]:
    """
    Mask sensitive headers for logging.
    
    Args:
        headers: Original headers dict
        
    Returns:
        New headers dict with sensitive values masked
    """
    if not headers:
        return {}
    
    masked = {}
    for key, value in headers.items():
        if is_sensitive_header(key):
            masked[key] = REDACTED_VALUE
        else:
            masked[key] = value
    
    return masked





def mask_url_credentials(url: str) -> str:
    """
    Mask any credentials embedded in a URL.
    
    Args:
        url: URL that may contain embedded credentials
        
    Returns:
        URL with credentials masked
    """
    # Match URLs with embedded credentials like http://user:pass@host
    pattern = r'(https?://)([^:]+):([^@]+)@'
    return re.sub(pattern, r'\1\2:' + REDACTED_VALUE + '@', url)





class AuditContext:
    """Context manager for audit logging of HTTP requests."""
    
    def __init__(
        self,
        method: str,
        url: str,
        headers: Optional[Dict[str, str]] = None,
    ):
        self.request_id = generate_request_id()
        self.method = method
        self.url = mask_url_credentials(url)
        self.headers = mask_headers(headers or {})
        # Store timestamps in milliseconds for granular timing
        self._start_time_ms: Optional[float] = None
        self._end_time_ms: Optional[float] = None
        self.status_code: Optional[int] = None
        self.error: Optional[str] = None
    
    def __enter__(self) -> "AuditContext":
        self._start_time_ms = time.perf_counter() * 1000
        logger.info(
            f"HTTP request started",
            extra={
                "request_id": self.request_id,
                "method": self.method,
                "url": self.url,
            }
        )
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb) -> None:
        self._end_time_ms = time.perf_counter() * 1000
        elapsed_ms = self._end_time_ms - self._start_time_ms
        
        log_data = {
            "request_id": self.request_id,
            "method": self.method,
            "url": self.url,
            "elapsed_ms": round(elapsed_ms, 2),
        }
        
        if self.status_code is not None:
            log_data["status_code"] = self.status_code
        
        if exc_type is not None:
            log_data["error"] = str(exc_val)
            logger.error("HTTP request failed", extra=log_data)
        else:
            logger.info("HTTP request completed", extra=log_data)
    


    def set_response(self, status_code: int):
        """Set the response status code for logging."""
        self.status_code = status_code

    
    @property
    def elapsed_ms(self) -> Optional[float]:
        """Get elapsed time in milliseconds (primary timing unit)."""
        if self._start_time_ms is None:
            return None
        end = self._end_time_ms or (time.perf_counter() * 1000)
        return end - self._start_time_ms
    
    @property
    def elapsed_seconds(self) -> Optional[float]:
        """Get elapsed time in seconds (derived from milliseconds)."""
        ms = self.elapsed_ms
        return ms / 1000 if ms is not None else None
    


    def to_dict(self) -> Dict[str, Any]:
        """
        Convert audit context to dictionary for structured logging.
        
        Returns:
            Dict with all audit fields
        """
        result = {
            "request_id": self.request_id,
            "method": self.method,
            "url": self.url,
            "headers": self.headers,
        }
        
        if self._start_time_ms:
            result["start_time_ms"] = self._start_time_ms
        if self._end_time_ms:
            result["end_time_ms"] = self._end_time_ms
            result["elapsed_ms"] = round(self._end_time_ms - self._start_time_ms, 2)
        if self.status_code:
            result["status_code"] = self.status_code
        if self.error:
            result["error"] = self.error
        
        return result