Auto-Generated API Reference¶

def load_settings() -> ApiSettings:
    """Load API settings from a config file and environment variables.

    Environment variables override config file values when present.
    """
    config_path = os.environ.get("FO_API_CONFIG_PATH")
    data: dict[str, Any] = {}

    if config_path:
        # Config file path is provided by deployment configuration, not request data.
        path = Path(config_path).expanduser()  # codeql[py/path-injection]
        if path.exists():
            payload = _load_yaml(path)
            data.update(payload.get("api", payload))
        else:
            logger.warning("API config path does not exist: {}", path)

    env = os.environ
    if "FO_API_APP_NAME" in env:
        data["app_name"] = env["FO_API_APP_NAME"]
    if "FO_API_VERSION" in env:
        data["version"] = env["FO_API_VERSION"]
    if "FO_API_ENVIRONMENT" in env:
        data["environment"] = env["FO_API_ENVIRONMENT"]
    if "FO_API_HOST" in env:
        data["host"] = env["FO_API_HOST"]
    if "FO_API_PORT" in env:
        try:
            data["port"] = int(env["FO_API_PORT"])
        except ValueError:
            logger.warning("Invalid FO_API_PORT value: {}", env["FO_API_PORT"])
    if "FO_API_LOG_LEVEL" in env:
        data["log_level"] = env["FO_API_LOG_LEVEL"]
    if "FO_API_CORS_ORIGINS" in env:
        data["cors_origins"] = _parse_list(env["FO_API_CORS_ORIGINS"])
    if "FO_API_CORS_ALLOW_METHODS" in env:
        data["cors_allow_methods"] = _parse_list(env["FO_API_CORS_ALLOW_METHODS"])
    if "FO_API_CORS_ALLOW_HEADERS" in env:
        data["cors_allow_headers"] = _parse_list(env["FO_API_CORS_ALLOW_HEADERS"])
    if "FO_API_CORS_ALLOW_CREDENTIALS" in env:
        data["cors_allow_credentials"] = env["FO_API_CORS_ALLOW_CREDENTIALS"].lower() in (
            "1",
            "true",
            "yes",
        )
    if "FO_API_ENABLE_DOCS" in env:
        data["enable_docs"] = env["FO_API_ENABLE_DOCS"].lower() in ("1", "true", "yes")
    if "FO_API_ALLOWED_PATHS" in env:
        data["allowed_paths"] = _parse_list(env["FO_API_ALLOWED_PATHS"])
    if "FO_API_WS_PING_INTERVAL" in env:
        try:
            interval = int(env["FO_API_WS_PING_INTERVAL"])
            if interval > 0:
                data["websocket_ping_interval"] = interval
            else:
                logger.warning(
                    "Invalid FO_API_WS_PING_INTERVAL value (must be > 0): {}",
                    env["FO_API_WS_PING_INTERVAL"],
                )
        except ValueError:
            logger.warning(
                "Invalid FO_API_WS_PING_INTERVAL value: {}",
                env["FO_API_WS_PING_INTERVAL"],
            )
    if "FO_API_WEBSOCKET_TOKEN" in env:
        data["websocket_token"] = env["FO_API_WEBSOCKET_TOKEN"]
    if "FO_API_AUTH_ENABLED" in env:
        data["auth_enabled"] = env["FO_API_AUTH_ENABLED"].lower() in ("1", "true", "yes")
    if "FO_API_AUTH_DB_PATH" in env:
        data["auth_db_path"] = env["FO_API_AUTH_DB_PATH"]
    if "FO_API_AUTH_JWT_SECRET" in env:
        data["auth_jwt_secret"] = env["FO_API_AUTH_JWT_SECRET"]
    if "FO_API_AUTH_JWT_ALGORITHM" in env:
        data["auth_jwt_algorithm"] = env["FO_API_AUTH_JWT_ALGORITHM"]
    if "FO_API_AUTH_ACCESS_MINUTES" in env:
        try:
            data["auth_access_token_minutes"] = int(env["FO_API_AUTH_ACCESS_MINUTES"])
        except ValueError:
            logger.warning(
                "Invalid FO_API_AUTH_ACCESS_MINUTES value: {}",
                env["FO_API_AUTH_ACCESS_MINUTES"],
            )
    if "FO_API_AUTH_REFRESH_DAYS" in env:
        try:
            data["auth_refresh_token_days"] = int(env["FO_API_AUTH_REFRESH_DAYS"])
        except ValueError:
            logger.warning(
                "Invalid FO_API_AUTH_REFRESH_DAYS value: {}",
                env["FO_API_AUTH_REFRESH_DAYS"],
            )
    if "FO_API_AUTH_REDIS_URL" in env:
        data["auth_redis_url"] = env["FO_API_AUTH_REDIS_URL"]
    elif "FO_REDIS_URL" in env:
        data["auth_redis_url"] = env["FO_REDIS_URL"]

    if "FO_API_AUTH_LOGIN_RATE_LIMIT" in env:
        data["auth_login_rate_limit_enabled"] = env["FO_API_AUTH_LOGIN_RATE_LIMIT"].lower() in (
            "1",
            "true",
            "yes",
        )
    if "FO_API_AUTH_LOGIN_MAX_ATTEMPTS" in env:
        try:
            data["auth_login_max_attempts"] = int(env["FO_API_AUTH_LOGIN_MAX_ATTEMPTS"])
        except ValueError:
            logger.warning(
                "Invalid FO_API_AUTH_LOGIN_MAX_ATTEMPTS value: {}",
                env["FO_API_AUTH_LOGIN_MAX_ATTEMPTS"],
            )
    if "FO_API_AUTH_LOGIN_WINDOW_SECONDS" in env:
        try:
            data["auth_login_window_seconds"] = int(env["FO_API_AUTH_LOGIN_WINDOW_SECONDS"])
        except ValueError:
            logger.warning(
                "Invalid FO_API_AUTH_LOGIN_WINDOW_SECONDS value: {}",
                env["FO_API_AUTH_LOGIN_WINDOW_SECONDS"],
            )
    if "FO_API_AUTH_PASSWORD_MIN_LENGTH" in env:
        try:
            data["auth_password_min_length"] = int(env["FO_API_AUTH_PASSWORD_MIN_LENGTH"])
        except ValueError:
            logger.warning(
                "Invalid FO_API_AUTH_PASSWORD_MIN_LENGTH value: {}",
                env["FO_API_AUTH_PASSWORD_MIN_LENGTH"],
            )
    if "FO_API_AUTH_PASSWORD_REQUIRE_NUMBER" in env:
        data["auth_password_require_number"] = env[
            "FO_API_AUTH_PASSWORD_REQUIRE_NUMBER"
        ].lower() in ("1", "true", "yes")
    if "FO_API_AUTH_PASSWORD_REQUIRE_LETTER" in env:
        data["auth_password_require_letter"] = env[
            "FO_API_AUTH_PASSWORD_REQUIRE_LETTER"
        ].lower() in ("1", "true", "yes")
    if "FO_API_AUTH_PASSWORD_REQUIRE_SPECIAL" in env:
        data["auth_password_require_special"] = env[
            "FO_API_AUTH_PASSWORD_REQUIRE_SPECIAL"
        ].lower() in ("1", "true", "yes")
    if "FO_API_AUTH_PASSWORD_REQUIRE_UPPERCASE" in env:
        data["auth_password_require_uppercase"] = env[
            "FO_API_AUTH_PASSWORD_REQUIRE_UPPERCASE"
        ].lower() in ("1", "true", "yes")
    if "FO_API_AUTH_BOOTSTRAP_ADMIN" in env:
        data["auth_bootstrap_admin"] = env["FO_API_AUTH_BOOTSTRAP_ADMIN"].lower() in (
            "1",
            "true",
            "yes",
        )
    if "FO_API_AUTH_BOOTSTRAP_LOCAL_ONLY" in env:
        data["auth_bootstrap_admin_local_only"] = env[
            "FO_API_AUTH_BOOTSTRAP_LOCAL_ONLY"
        ].lower() in ("1", "true", "yes")
    if "FO_API_DATABASE_URL" in env:
        data["database_url"] = env["FO_API_DATABASE_URL"]
    if "FO_API_DB_POOL_SIZE" in env:
        try:
            data["database_pool_size"] = int(env["FO_API_DB_POOL_SIZE"])
        except ValueError:
            logger.warning("Invalid FO_API_DB_POOL_SIZE value: {}", env["FO_API_DB_POOL_SIZE"])
    if "FO_API_DB_MAX_OVERFLOW" in env:
        try:
            data["database_max_overflow"] = int(env["FO_API_DB_MAX_OVERFLOW"])
        except ValueError:
            logger.warning(
                "Invalid FO_API_DB_MAX_OVERFLOW value: {}",
                env["FO_API_DB_MAX_OVERFLOW"],
            )
    if "FO_API_DB_POOL_PRE_PING" in env:
        data["database_pool_pre_ping"] = env["FO_API_DB_POOL_PRE_PING"].lower() in (
            "1",
            "true",
            "yes",
        )
    if "FO_API_DB_POOL_RECYCLE_SECONDS" in env:
        try:
            data["database_pool_recycle_seconds"] = int(env["FO_API_DB_POOL_RECYCLE_SECONDS"])
        except ValueError:
            logger.warning(
                "Invalid FO_API_DB_POOL_RECYCLE_SECONDS value: {}",
                env["FO_API_DB_POOL_RECYCLE_SECONDS"],
            )
    if "FO_API_DB_ECHO" in env:
        data["database_echo"] = env["FO_API_DB_ECHO"].lower() in ("1", "true", "yes")
    if "FO_API_CACHE_REDIS_URL" in env:
        data["cache_redis_url"] = env["FO_API_CACHE_REDIS_URL"]
    elif "FO_REDIS_URL" in env:
        data["cache_redis_url"] = env["FO_REDIS_URL"]
    if "FO_API_CACHE_TTL_SECONDS" in env:
        try:
            data["cache_default_ttl_seconds"] = int(env["FO_API_CACHE_TTL_SECONDS"])
        except ValueError:
            logger.warning(
                "Invalid FO_API_CACHE_TTL_SECONDS value: {}",
                env["FO_API_CACHE_TTL_SECONDS"],
            )
    if "FO_API_API_KEY_ENABLED" in env:
        data["api_key_enabled"] = env["FO_API_API_KEY_ENABLED"].lower() in ("1", "true", "yes")
    if "FO_API_API_KEY_ADMIN" in env:
        data["api_key_admin"] = env["FO_API_API_KEY_ADMIN"].lower() in ("1", "true", "yes")
    if "FO_API_API_KEY_HEADER" in env:
        data["api_key_header"] = env["FO_API_API_KEY_HEADER"]
    if "FO_API_API_KEYS" in env:
        raw_keys = _parse_list(env["FO_API_API_KEYS"])
        data["api_key_hashes"] = [hash_api_key(key) for key in raw_keys]
    if "FO_API_API_KEY_HASHES" in env:
        data["api_key_hashes"] = _parse_list(env["FO_API_API_KEY_HASHES"])
    if "FO_API_RATE_LIMIT_ENABLED" in env:
        data["rate_limit_enabled"] = env["FO_API_RATE_LIMIT_ENABLED"].lower() in (
            "1",
            "true",
            "yes",
        )
    if "FO_API_RATE_LIMIT_DEFAULT_REQUESTS" in env:
        try:
            data["rate_limit_default_requests"] = int(env["FO_API_RATE_LIMIT_DEFAULT_REQUESTS"])
        except ValueError:
            logger.warning(
                "Invalid FO_API_RATE_LIMIT_DEFAULT_REQUESTS value: {}",
                env["FO_API_RATE_LIMIT_DEFAULT_REQUESTS"],
            )
    if "FO_API_RATE_LIMIT_DEFAULT_WINDOW_SECONDS" in env:
        try:
            data["rate_limit_default_window_seconds"] = int(
                env["FO_API_RATE_LIMIT_DEFAULT_WINDOW_SECONDS"]
            )
        except ValueError:
            logger.warning(
                "Invalid FO_API_RATE_LIMIT_DEFAULT_WINDOW_SECONDS value: {}",
                env["FO_API_RATE_LIMIT_DEFAULT_WINDOW_SECONDS"],
            )
    if "FO_API_RATE_LIMIT_TRUST_PROXY_HEADERS" in env:
        data["rate_limit_trust_proxy_headers"] = env[
            "FO_API_RATE_LIMIT_TRUST_PROXY_HEADERS"
        ].lower() in ("1", "true", "yes")
    if "FO_API_RATE_LIMIT_EXEMPT_PATHS" in env:
        data["rate_limit_exempt_paths"] = _parse_list(env["FO_API_RATE_LIMIT_EXEMPT_PATHS"])
    if "FO_API_RATE_LIMIT_RULES" in env:
        try:
            parsed_rules = json.loads(env["FO_API_RATE_LIMIT_RULES"])
            if isinstance(parsed_rules, dict):
                data["rate_limit_rules"] = parsed_rules
        except json.JSONDecodeError:
            logger.warning("Invalid FO_API_RATE_LIMIT_RULES JSON value")
    if "FO_API_SECURITY_HEADERS_ENABLED" in env:
        data["security_headers_enabled"] = env["FO_API_SECURITY_HEADERS_ENABLED"].lower() in (
            "1",
            "true",
            "yes",
        )
    if "FO_API_SECURITY_CSP" in env:
        data["security_csp"] = env["FO_API_SECURITY_CSP"]
    if "FO_API_SECURITY_HSTS_SECONDS" in env:
        try:
            data["security_hsts_seconds"] = int(env["FO_API_SECURITY_HSTS_SECONDS"])
        except ValueError:
            logger.warning(
                "Invalid FO_API_SECURITY_HSTS_SECONDS value: {}",
                env["FO_API_SECURITY_HSTS_SECONDS"],
            )
    if "FO_API_SECURITY_HSTS_SUBDOMAINS" in env:
        data["security_hsts_subdomains"] = env["FO_API_SECURITY_HSTS_SUBDOMAINS"].lower() in (
            "1",
            "true",
            "yes",
        )
    if "FO_API_SECURITY_REFERRER_POLICY" in env:
        data["security_referrer_policy"] = env["FO_API_SECURITY_REFERRER_POLICY"]
    if "FO_OLLAMA_URL" in env:
        data["ollama_url"] = env["FO_OLLAMA_URL"]
    elif "OLLAMA_HOST" in env:
        data["ollama_url"] = env["OLLAMA_HOST"]

    api_key_enabled_explicit = "api_key_enabled" in data
    settings = ApiSettings(**data)
    if settings.auth_enabled and settings.auth_jwt_secret.get_secret_value() == "change-me":
        if settings.environment.lower() in {"development", "test"}:
            logger.warning(
                "FO_API_AUTH_JWT_SECRET is using the default placeholder. "
                "Set FO_API_AUTH_JWT_SECRET before deploying."
            )
        else:
            raise ValueError(
                "FO_API_AUTH_JWT_SECRET must be set when auth is enabled outside development."
            )
    if settings.api_key_enabled and not settings.api_key_hashes and api_key_enabled_explicit:
        logger.warning("API key auth is enabled but no keys are configured.")
    if settings.environment.lower() not in {"development", "test"}:
        if "*" in settings.cors_origins:
            raise ValueError("CORS origins must be explicit in production.")
        if any("localhost" in origin or "127.0.0.1" in origin for origin in settings.cors_origins):
            raise ValueError("Localhost CORS origins must be removed in production.")
    return settings

def verify_password(plain_password: str, hashed_password: str) -> bool:
    """Return True if plain_password matches hashed_password."""
    return bool(_PWD_CONTEXT.verify(plain_password, hashed_password))

def hash_password(password: str) -> str:
    """Return the bcrypt hash of password."""
    return str(_PWD_CONTEXT.hash(password))

def validate_password(password: str, settings: ApiSettings) -> tuple[bool, str]:
    """Validate password strength based on API settings."""
    if len(password) < settings.auth_password_min_length:
        return (
            False,
            f"Password must be at least {settings.auth_password_min_length} characters long",
        )
    if settings.auth_password_require_letter and not any(ch.isalpha() for ch in password):
        return False, "Password must include at least one letter"
    if settings.auth_password_require_number and not any(ch.isdigit() for ch in password):
        return False, "Password must include at least one number"
    if settings.auth_password_require_uppercase and not any(ch.isupper() for ch in password):
        return False, "Password must include at least one uppercase letter"
    if settings.auth_password_require_special and not any(
        ch in "!\"#$%&'()*+,-./:;<=>?@[\\]^_`{|}~" for ch in password
    ):
        return False, "Password must include at least one special character"
    if password.lower() in _COMMON_PASSWORDS:
        return False, "Password is too common, please choose a more unique password"
    return True, ""

def create_token_bundle(user_id: str, username: str, settings: ApiSettings) -> TokenBundle:
    """Create a new access and refresh token bundle for a user."""
    payload = {"sub": username, "user_id": user_id}
    access_token, access_jti, access_exp = _build_token(
        payload,
        _ACCESS_TOKEN_TYPE,
        timedelta(minutes=settings.auth_access_token_minutes),
        settings,
    )
    refresh_token, refresh_jti, refresh_exp = _build_token(
        payload,
        _REFRESH_TOKEN_TYPE,
        timedelta(days=settings.auth_refresh_token_days),
        settings,
    )
    return TokenBundle(
        access_token=access_token,
        refresh_token=refresh_token,
        access_jti=access_jti,
        refresh_jti=refresh_jti,
        access_expires_at=access_exp,
        refresh_expires_at=refresh_exp,
    )

def decode_token(token: str, settings: ApiSettings) -> dict[str, Any]:
    """Decode and return the JWT payload, raising TokenError if invalid."""
    try:
        result: dict[str, Any] = jwt.decode(
            token,
            settings.auth_jwt_secret.get_secret_value(),
            algorithms=[settings.auth_jwt_algorithm],
        )
        return result
    except JWTError as exc:
        raise TokenError("Invalid token") from exc

def is_access_token(payload: dict[str, Any]) -> bool:
    """Return True if payload represents an access token."""
    return payload.get("type") == _ACCESS_TOKEN_TYPE

def is_refresh_token(payload: dict[str, Any]) -> bool:
    """Return True if payload represents a refresh token."""
    return payload.get("type") == _REFRESH_TOKEN_TYPE

def generate_api_key(prefix: str = "fo") -> str:
    """Generate a new API key."""
    key_id = secrets.token_hex(4)
    token = secrets.token_urlsafe(32)
    return f"{prefix}_{key_id}_{token}"

def hash_api_key(api_key: str) -> str:
    """Hash an API key for storage."""
    return cast(str, _API_KEY_CONTEXT.hash(api_key))

def match_api_key_hash(api_key: str, hashes: Iterable[str]) -> str | None:
    """Return the stored hash matching an API key, if any."""
    for stored in hashes:
        try:
            if _API_KEY_CONTEXT.verify(api_key, stored):
                return stored
        except (ValueError, TypeError):
            continue
    return None

def verify_api_key(api_key: str, hashes: Iterable[str]) -> bool:
    """Verify an API key against stored hashes."""
    return match_api_key_hash(api_key, hashes) is not None

def api_key_identifier(api_key: str, hashes: Iterable[str]) -> str | None:
    """Return a stable identifier derived from the stored hash."""
    matched = match_api_key_hash(api_key, hashes)
    if not matched:
        return None
    parts = api_key.split("_", 2)
    if len(parts) == 3 and parts[1]:
        return parts[1]
    return matched[-12:]

def build_rate_limiter(redis_url: Optional[str]) -> RateLimiter:
    """Create a rate limiter instance."""
    if not redis_url:
        return InMemoryRateLimiter()
    try:
        client = Redis.from_url(redis_url, decode_responses=True)
        client.ping()
        return RedisRateLimiter(client)
    except Exception as exc:
        logger.warning("Rate limiter Redis unavailable, using in-memory limiter: {}", exc)
        return InMemoryRateLimiter()

def build_login_rate_limiter(
    redis_url: Optional[str],
    max_attempts: int,
    window_seconds: int,
) -> LoginRateLimiter:
    """Create a login rate limiter, preferring Redis when configured."""
    if not redis_url:
        return InMemoryLoginRateLimiter(max_attempts=max_attempts, window_seconds=window_seconds)
    try:
        client = Redis.from_url(redis_url, decode_responses=True)
        client.ping()
        return RedisLoginRateLimiter(
            redis=client,
            max_attempts=max_attempts,
            window_seconds=window_seconds,
        )
    except Exception as exc:
        logger.warning("Auth redis unavailable, using in-memory rate limiter: {}", exc)
        return InMemoryLoginRateLimiter(max_attempts=max_attempts, window_seconds=window_seconds)

def build_token_store(redis_url: Optional[str]) -> TokenStore:
    """Create a token store, preferring Redis when configured."""
    if not redis_url:
        return InMemoryTokenStore()
    try:
        client = Redis.from_url(redis_url, decode_responses=True)
        client.ping()
        return RedisTokenStore(client)
    except Exception as exc:
        logger.warning("Auth redis unavailable, using in-memory token store: {}", exc)
        return InMemoryTokenStore()