"""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: :<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 ':' 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