validators.py

"""Input validators for hosting-panel form fields.

Centralises format checks so every route that accepts a token / slug /
secret gets the same defence. Each validator returns the sanitised
value on success or raises HTTPException on failure — callers in
routes.py chain these via `Form(...) = Depends(...)` patterns or by
calling them directly.

Defence notes:
- All lengths are bounded so a multi-MB `slug` cannot DoS the DB layer
  or push huge strings into HF Space names (HF rejects names > 32 chars
  but we cap earlier at 64 to be safe).
- We don't allow path separators, whitespace, or `..` in slugs —
  HuggingFace space names share a single DNS-like namespace and a
  malicious slug could collide with another user.
- Tokens are length-checked against their format. A bot_token that's
  4 chars is a junk submission; 5000 chars is a DoS attempt.
"""

from __future__ import annotations

import re

from fastapi import HTTPException, status


# ---- Format patterns -----------------------------------------------------

_SLUG_RE = re.compile(r"^[a-z0-9][a-z0-9_-]{1,62}[a-z0-9]$")
_HF_SPACE_RE = re.compile(r"^[a-zA-Z0-9][a-zA-Z0-9_-]{1,62}[a-zA-Z0-9]$")
_HF_DATASET_RE = re.compile(r"^[a-zA-Z0-9_.-]+/[a-zA-Z0-9_.-]+$")
_BOT_TOKEN_LEN = (35, 60)         # Telegram bot tokens: 35 chars + "bot" prefix allowed variants
_HF_TOKEN_LEN = (30, 60)          # hf_xxxxx... typically 37 chars
_PROXY_TOKEN_LEN = (16, 128)      # random shared secret; allow wide range
_NVIDIA_KEY_LEN = (20, 200)       # nvapi-... typically ~60 chars
_MODEL_RE = re.compile(r"^[A-Za-z0-9_./-]+$")


# ---- Public validators ---------------------------------------------------

def validate_slug(raw: str) -> str:
    """Strict lowercase slug — used for `BotInstance.slug` and HF space names.

    Rejects empty, leading/trailing hyphens, double hyphens, uppercase,
    path separators, and anything that doesn't match `[a-z0-9][a-z0-9_-]+[a-z0-9]`.
    """
    if not raw or not isinstance(raw, str):
        raise HTTPException(status_code=400, detail="slug is required")
    slug = raw.strip().lower()
    if not _SLUG_RE.match(slug):
        raise HTTPException(
            status_code=400,
            detail=(
                "slug must be 3-64 chars, lowercase, start/end with alnum, "
                "and contain only a-z, 0-9, '_' or '-'"
            ),
        )
    return slug


def validate_hf_space_name(raw: str) -> str:
    """HF space name — slightly looser (allows uppercase) but still strict."""
    if not raw or not isinstance(raw, str):
        raise HTTPException(status_code=400, detail="hf_space_name is required")
    name = raw.strip()
    if not _HF_SPACE_RE.match(name):
        raise HTTPException(
            status_code=400,
            detail="hf_space_name must be 3-64 alphanumeric/underscore/hyphen chars",
        )
    return name


def validate_hf_dataset_repo(raw: str) -> str:
    """HF dataset repo: must be `owner/name`."""
    if not raw or not isinstance(raw, str):
        raise HTTPException(status_code=400, detail="hf_dataset_repo is required")
    repo = raw.strip()
    if not _HF_DATASET_RE.match(repo) or repo.count("/") != 1:
        raise HTTPException(
            status_code=400,
            detail="hf_dataset_repo must be in 'owner/name' format",
        )
    return repo


def validate_bot_token(raw: str) -> str:
    """Telegram bot token. Format: <bot_id>:<44-char base64-ish secret>."""
    if not raw or not isinstance(raw, str):
        raise HTTPException(status_code=400, detail="bot_token is required")
    tok = raw.strip()
    lo, hi = _BOT_TOKEN_LEN
    if not (lo <= len(tok) <= hi):
        raise HTTPException(
            status_code=400,
            detail=f"bot_token length must be {lo}-{hi} chars",
        )
    if ":" not in tok:
        raise HTTPException(
            status_code=400,
            detail="bot_token must be in '<bot_id>:<secret>' format",
        )
    # Defensive: reject whitespace and control chars inside the token.
    if any(c.isspace() or ord(c) < 32 for c in tok):
        raise HTTPException(status_code=400, detail="bot_token contains invalid chars")
    return tok


def validate_hf_token(raw: str) -> str:
    """HuggingFace user access token (`hf_...`)."""
    if not raw or not isinstance(raw, str):
        raise HTTPException(status_code=400, detail="hf_token is required")
    tok = raw.strip()
    lo, hi = _HF_TOKEN_LEN
    if not (lo <= len(tok) <= hi):
        raise HTTPException(
            status_code=400,
            detail=f"hf_token length must be {lo}-{hi} chars",
        )
    if any(c.isspace() or ord(c) < 32 for c in tok):
        raise HTTPException(status_code=400, detail="hf_token contains invalid chars")
    return tok


def validate_proxy_token(raw: str) -> str:
    """Shared RENDER/LLM proxy secret."""
    if not raw or not isinstance(raw, str):
        raise HTTPException(status_code=400, detail="llm_proxy_token is required")
    tok = raw.strip()
    lo, hi = _PROXY_TOKEN_LEN
    if not (lo <= len(tok) <= hi):
        raise HTTPException(
            status_code=400,
            detail=f"llm_proxy_token length must be {lo}-{hi} chars",
        )
    if any(c.isspace() or ord(c) < 32 for c in tok):
        raise HTTPException(status_code=400, detail="llm_proxy_token contains invalid chars")
    return tok


def validate_nvidia_api_key(raw: str) -> str:
    """NVIDIA API key (currently `nvapi-...`)."""
    if not raw or not isinstance(raw, str):
        raise HTTPException(status_code=400, detail="nvidia_api_key is required")
    key = raw.strip()
    lo, hi = _NVIDIA_KEY_LEN
    if not (lo <= len(key) <= hi):
        raise HTTPException(
            status_code=400,
            detail=f"nvidia_api_key length must be {lo}-{hi} chars",
        )
    if any(c.isspace() or ord(c) < 32 for c in key):
        raise HTTPException(status_code=400, detail="nvidia_api_key contains invalid chars")
    return key


def validate_optional_nvidia_model(raw: str) -> str:
    """NVIDIA model id — e.g. `minimaxai/minimax-m3`."""
    if not raw:
        return "minimaxai/minimax-m3"
    model = raw.strip()
    if not _MODEL_RE.match(model):
        raise HTTPException(
            status_code=400,
            detail="nvidia_model must contain only letters, digits, '_', '-', '.', '/'",
        )
    return model


def validate_optional_url(raw: str, field: str) -> str:
    """Optional URL field — empty allowed; otherwise basic sanity."""
    if not raw:
        return ""
    url = raw.strip()
    if not (url.startswith("http://") or url.startswith("https://")):
        raise HTTPException(status_code=400, detail=f"{field} must start with http:// or https://")
    if len(url) > 2048:
        raise HTTPException(status_code=400, detail=f"{field} too long (max 2048 chars)")
    return url


def validate_source_repo(raw: str, source_type: str = "hf") -> str:
    """Bot's source code location.

    ``hf``     → ``owner/name`` (HF Hub repo)
    ``github`` → ``https://github.com/owner/repo[.git]`` or any cloneable URL
    ``local``  → optional marker; the panel keeps the zip on disk instead.

    ``local`` accepts an empty string (no external repo).  ``hf`` and
    ``github`` still require a non-empty value.
    """
    if source_type == "local":
        # Local bots get their source from an uploaded zip, not an
        # external repo.  An empty string is fine — the panel will
        # store the zip path on the BotInstance row instead.
        return ""
    if not raw or not isinstance(raw, str):
        raise HTTPException(status_code=400, detail="source_repo is required")
    repo = raw.strip()
    if source_type == "github":
        if not (repo.startswith("http://") or repo.startswith("https://") or repo.startswith("git@")):
            raise HTTPException(
                status_code=400,
                detail="source_repo must be a git URL (https://… or git@…) when source_type=github",
            )
        if len(repo) > 256:
            raise HTTPException(status_code=400, detail="source_repo too long (max 256 chars)")
        return repo
    # Default: HF Hub ``owner/name``.
    if "/" not in repo or repo.count("/") > 1:
        raise HTTPException(
            status_code=400,
            detail="source_repo must be in 'owner/name' format for HF Hub",
        )
    if not _HF_DATASET_RE.match(repo):
        raise HTTPException(
            status_code=400,
            detail="source_repo must match 'owner/name' (alphanumeric, '-', '_', '.')",
        )
    if len(repo) > 256:
        raise HTTPException(status_code=400, detail="source_repo too long (max 256 chars)")
    return repo


def validate_source_type(raw: str) -> str:
    """``hf``, ``github`` or ``local``.

    ``local`` means the bot's source lives as a zip uploaded through
    the create form (or replaced later via the Files tab).  Deploys
    from a ``local`` source aren't wired up to DeployOrchestrator yet
    — the bot stays in ``pending`` until the user wires source on the
    bot detail page.
    """
    val = (raw or "hf").strip().lower()
    if val not in ("hf", "github", "local"):
        raise HTTPException(
            status_code=400,
            detail="source_type must be 'hf', 'github' or 'local'",
        )
    return val


def validate_framework(raw: str) -> str:
    """Free-form framework hint (e.g. ``docker``, ``python``, ``node``)."""
    if not raw:
        return "docker"
    val = raw.strip().lower()
    if len(val) > 64:
        raise HTTPException(status_code=400, detail="framework too long (max 64 chars)")
    if not re.match(r"^[a-z0-9_./-]+$", val):
        raise HTTPException(
            status_code=400,
            detail="framework must be lowercase alphanumerics with _-./",
        )
    return val


def validate_branch(raw: str) -> str:
    """Git branch name. Conservative whitelist."""
    if not raw:
        return "main"
    val = raw.strip()
    if len(val) > 64:
        raise HTTPException(status_code=400, detail="branch name too long (max 64 chars)")
    if not re.match(r"^[A-Za-z0-9_./-]+$", val):
        raise HTTPException(
            status_code=400,
            detail="branch may only contain alphanumerics, '.', '_', '-', '/'",
        )
    return val