Data Protection¶

# From backend/api/routes/media.py:42-124
def _validate_and_resolve_path(base_path: Path, requested_path: str) -> Path:
    """Validate and resolve a file path securely."""

    # Check path length to prevent buffer overflow attacks
    if len(requested_path) > MAX_PATH_LENGTH:
        raise HTTPException(
            status_code=status.HTTP_414_URI_TOO_LONG,
            detail=MediaErrorResponse(
                error=f"Path too long. Maximum length is {MAX_PATH_LENGTH} characters.",
                path=requested_path[:100] + "...",
            ).model_dump(),
        )

    # Check for path traversal attempts
    if ".." in requested_path or requested_path.startswith("/"):
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail=MediaErrorResponse(
                error="Path traversal detected",
                path=requested_path,
            ).model_dump(),
        )

    # Resolve the full path
    full_path = (base_path / requested_path).resolve()

    # Ensure the resolved path is within the base directory
    try:
        full_path.relative_to(base_path.resolve())
    except ValueError as err:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail=MediaErrorResponse(
                error="Access denied - path outside allowed directory",
                path=requested_path,
            ).model_dump(),
        ) from err

    return full_path

# From backend/api/routes/media.py:31-39
ALLOWED_TYPES = {
    ".jpg": "image/jpeg",
    ".jpeg": "image/jpeg",
    ".png": "image/png",
    ".gif": "image/gif",
    ".mp4": "video/mp4",
    ".avi": "video/x-msvideo",
    ".webm": "video/webm",
}

/cameras/
  ├── front_door/
  │   ├── 2024/
  │   │   └── 01/
  │   │       └── 15/
  │   │           └── image_001.jpg
  ├── backyard/
  │   └── ...

# From backend/api/middleware/auth.py (conceptual)
import hashlib

def hash_api_key(api_key: str) -> str:
    """Hash an API key using SHA-256."""
    return hashlib.sha256(api_key.encode()).hexdigest()

def verify_api_key(provided_key: str, stored_hash: str) -> bool:
    """Verify an API key against its stored hash."""
    return hash_api_key(provided_key) == stored_hash

# From backend/core/config.py:1090-1098
api_key_enabled: bool = Field(
    default=False,
    description="Enable API key authentication (default: False for development)",
)
api_keys: list[str] = Field(
    default=[],
    description="List of valid API keys (plain text, hashed on startup)",
)

# From backend/core/sanitization.py:133-230
def sanitize_error_for_response(error: Exception, context: str = "") -> str:
    """Sanitize an exception for safe inclusion in API responses.

    Removes potentially sensitive information like:
    - Full file paths (keeps only filename)
    - Stack traces
    - Internal module names
    - Database connection details
    - URL credentials (user:password@host)
    - JSON password values
    - Windows paths
    """

# From backend/core/sanitization.py:200-217
sensitive_patterns = [
    # JSON-style password values
    (re.compile(r'(["\'])password\1\s*:\s*["\'][^"\']*["\']', re.IGNORECASE),
     '"password": "[REDACTED]"'),
    # Key-value password patterns
    (re.compile(r"password[=:]\s*\S+", re.IGNORECASE), "password=[REDACTED]"),
    # Bearer tokens
    (re.compile(r"Bearer\s+\S+", re.IGNORECASE), "Bearer [REDACTED]"),
    # API keys
    (re.compile(r"api[_-]?key[=:]\s*\S+", re.IGNORECASE), "api_key=[REDACTED]"),
    # Secret/token values
    (re.compile(r"secret[=:]\s*\S+", re.IGNORECASE), "secret=[REDACTED]"),
    (re.compile(r"token[=:]\s*\S+", re.IGNORECASE), "token=[REDACTED]"),
    # IPv4 addresses
    (re.compile(r"\b\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}\b"), "[IP_REDACTED]"),
]

# From backend/core/sanitization.py:103-130
def sanitize_path_for_error(path: str) -> str:
    """Sanitize a file path for safe inclusion in error messages.

    Removes the directory path and returns only the filename.
    """
    if not path:
        return "[unknown]"

    # Extract just the filename
    parts = path.replace("\\", "/").rsplit("/", 1)
    filename = parts[1] if len(parts) == 2 else parts[0]

    # Limit length
    if len(filename) > 100:
        filename = filename[:97] + "..."

    return filename

# From backend/api/middleware/request_recorder.py
def redact_request_body(body: dict) -> dict:
    """Redact sensitive fields from request body for logging."""
    sensitive_keys = {"password", "api_key", "secret", "token", "authorization"}
    redacted = {}
    for key, value in body.items():
        if key.lower() in sensitive_keys:
            redacted[key] = "[REDACTED]"
        elif isinstance(value, dict):
            redacted[key] = redact_request_body(value)
        else:
            redacted[key] = value
    return redacted

# From backend/core/config.py:618-623
retention_days: int = Field(
    default=30,
    gt=0,
    description="Number of days to retain events and detections",
)

# Cleanup removes:
# - Detection records older than retention_days
# - Event records older than retention_days
# - Orphaned media files
# - Expired Redis keys

# From backend/core/sanitization.py:38-77
def sanitize_container_name(name: str) -> str:
    """Sanitize a container name to prevent command injection.

    Only allows alphanumeric characters, hyphens, and underscores.
    Names must start with an alphanumeric character.
    """
    if not name:
        raise ValueError("Container name cannot be empty")

    if len(name) > CONTAINER_NAME_MAX_LENGTH:
        raise ValueError(f"Container name exceeds maximum length of {CONTAINER_NAME_MAX_LENGTH}")

    if not CONTAINER_NAME_PATTERN.match(name):
        raise ValueError("Container name contains invalid characters")

    return name

# Environment variable: DATABASE_URL
# Format: postgresql+asyncpg://user:password@host:port/database  # pragma: allowlist secret
# Password portion is never included in logs

# Safe - SQLAlchemy parameterizes all values
result = await session.execute(
    select(Detection).where(Detection.camera_id == camera_id)
)