backend/provisioning/space_provisioner.py

"""
Per-subscriber HF Space provisioning, with per-subscriber credential isolation.

This exists because `huggingface-spaces-hosting.md:48` requires it in as many words:
"Each subscriber's own Space needs its own credential/vault isolation, consistent with
credentials-and-secrets.md's 'every subscriber's provisioned credential must be
genuinely unique to them' — a per-user Space is exactly the kind of boundary that should
carry per-user credentials, not a shared set."

Design constraints taken directly from the two reference docs, not invented here:

* One shared, namespaced credential scheme — NOT a bespoke store per feature or per user
  (`credentials-and-secrets.md:43`). Every per-user secret is keyed by (purpose, user_id)
  through one derivation function, so adding a new secret type does not add a new system.
* Genuine per-user uniqueness with no collisions, verified rather than assumed
  (`credentials-and-secrets.md:45`). Two subscribers must never be able to alias to,
  overwrite, or be confused with each other's credentials.
* Clever, not brute-force (`credentials-and-secrets.md:41`). We do NOT register a new
  upstream account with every provider for every subscriber. Each subscriber gets a
  derived, individually-attributable token scoped to them, plus their own Space secrets;
  upstream provider capacity stays shared, but every call is attributable to exactly one
  account, which is what metering and billing actually require.
* The owner is exempt (`huggingface-spaces-hosting.md:51`, `credentials-and-secrets.md:53`).
  The hidden owner plan keeps its own existing hosting and is never provisioned a
  per-subscriber Space.

Derivation
----------
Per-user secrets are HKDF-style derived from one high-entropy root
(`SUBSCRIBER_ROOT_SECRET`) plus the user id and the purpose label. That gives:

  - uniqueness per (user, purpose) without storing N×M random values,
  - collision-resistance from SHA-256 over a length-prefixed, unambiguous input,
  - reproducibility, so a lost Space can be re-provisioned identically,
  - and revocability per user via a per-user `version` counter.

Length-prefixing matters: naively concatenating user id and purpose would let
("ab", "c") and ("a", "bc") derive the same key. The encoder below makes that
impossible, and `verify_no_collisions()` proves it rather than asserting it.

Nothing in this module creates a Space unless `dry_run=False` is passed explicitly —
provisioning a real Space is an outward-facing action and is left to an explicit,
authorised call.
"""

from __future__ import annotations

import hashlib
import hmac
import logging
import os
import secrets
from dataclasses import dataclass, field
from typing import Any

log = logging.getLogger(__name__)

# The purposes a subscriber's own Space needs its own value for. One namespaced scheme,
# per credentials-and-secrets.md:43 — adding a purpose here is the ONLY thing needed to
# give every subscriber their own isolated value for it.
#
# v22-2: these are the SYSTEM-DERIVED secrets. Every one is a value this system invents
# for itself, so HMAC derivation from a root secret works: the value has no meaning to
# any third party and nobody outside has to agree to it.
DERIVED_SECRET_PURPOSES: tuple[str, ...] = (
    "JARVIS_CLOUD_TOKEN",      # the subscriber's own API/WS bearer for their Space
    "VAULT_MASTER_PASSWORD",   # derives that Space's vault key; never shared
    "PBKDF2_SALT",             # per-subscriber salt, so no two vaults share a derivation
    "WEBHOOK_SIGNING_SECRET",  # payment-webhook HMAC, scoped to this subscriber
)

# v22-2: OWNER-SUPPLIED per-subscriber credentials — a genuinely different provisioning
# pattern the derivation scheme above cannot express, and previously had no place at all.
#
# The distinction that matters: a derived secret is one WE choose, so we can compute it.
# An owner-supplied secret is issued by a THIRD PARTY and the third party must agree it
# is valid. No amount of HMAC produces a Sketchfab key that Sketchfab will honour. These
# can therefore only ever be STORED AND LOOKED UP, never computed — which means
# provisioning has to be able to *block* on one being absent instead of silently
# producing a Space that 401s the first time a subscriber opens the model browser.
#
# Each entry: purpose -> (human description, where the owner obtains it).
OWNER_SUPPLIED_PURPOSES: dict[str, tuple[str, str]] = {
    "SKETCHFAB_API_KEY": (
        "That subscriber's own Sketchfab API token, used for model search/download in "
        "the AR connector drawer.",
        "The owner creates a separate Sketchfab account/token per subscriber at "
        "https://sketchfab.com/settings/password and records it against that user id.",
    ),
}

# Backwards-compatible alias. Everything that existed before v22-2 (verify_no_collisions,
# rotate_only_this_user, the Part 29 collision proof) means the derived set specifically,
# so this name keeps pointing at exactly what it always pointed at rather than silently
# growing to include credentials that cannot be derived.
SUBSCRIBER_SECRET_PURPOSES: tuple[str, ...] = DERIVED_SECRET_PURPOSES

OWNER_USER_ID = "owner"


def _root_secret() -> bytes:
    """The single high-entropy root every per-user secret is derived from.

    Sourced from the environment, never hardcoded — per credentials-and-secrets.md:22,
    an encryption/derivation root that ships inside the artifact protects nothing.
    """
    raw = os.environ.get("SUBSCRIBER_ROOT_SECRET", "").strip()
    if not raw:
        raise RuntimeError(
            "SUBSCRIBER_ROOT_SECRET is not set. Per-subscriber credentials cannot be "
            "derived without it. Set it as a Space Secret / environment variable; do "
            "not hardcode it, and do not fall back to a default — a predictable root "
            "would make every subscriber's credentials predictable."
        )
    return raw.encode("utf-8")


def _encode(*parts: str) -> bytes:
    """Unambiguous, length-prefixed encoding.

    Without this, derive("ab", "c") and derive("a", "bc") would hash identical bytes and
    two different subscribers could collide. Each part is prefixed with its byte length.
    """
    out = bytearray()
    for p in parts:
        b = p.encode("utf-8")
        out += len(b).to_bytes(4, "big") + b
    return bytes(out)


def derive_secret(user_id: str, purpose: str, version: int = 1) -> str:
    """Derive this subscriber's value for one purpose. Deterministic and unique."""
    if not user_id or not user_id.strip():
        raise ValueError("user_id is required")
    if purpose not in SUBSCRIBER_SECRET_PURPOSES:
        raise ValueError(f"unknown purpose {purpose!r}")
    mac = hmac.new(_root_secret(), _encode(user_id.strip(), purpose, str(version)), hashlib.sha256)
    # urlsafe, no padding — safe in env vars, headers and URLs alike.
    import base64
    return base64.urlsafe_b64encode(mac.digest()).decode().rstrip("=")


def subscriber_secrets(user_id: str, version: int = 1) -> dict[str, str]:
    """Every isolated secret for one subscriber's own Space."""
    return {p: derive_secret(user_id, p, version) for p in SUBSCRIBER_SECRET_PURPOSES}


# ─────────────────────────────────────────────────────────────────────────────
# v22-2 — owner-supplied per-subscriber credentials
#
# These route through the SAME vault the rest of the system uses, namespaced per
# (user, purpose), rather than a new bespoke store. credentials-and-secrets.md:43 is
# explicit about this: "resist the pull to bolt on a bespoke storage/rotation mechanism
# for each new kind of secret — extend the existing vault/credential-management path
# with a clean per-user namespace." A parallel store for one connector is exactly the
# unauditable-mess pattern that rule exists to prevent.
# ─────────────────────────────────────────────────────────────────────────────

def owner_supplied_vault_key(user_id: str, purpose: str) -> str:
    """The vault key one subscriber's owner-supplied credential is stored under.

    Namespaced by both user and purpose so two subscribers' Sketchfab keys can never
    alias to each other, per the no-collisions requirement that governs the derived
    secrets too.
    """
    if purpose not in OWNER_SUPPLIED_PURPOSES:
        raise ValueError(f"{purpose!r} is not an owner-supplied purpose")
    if not user_id or not user_id.strip():
        raise ValueError("user_id is required")
    return f"SUBSCRIBER::{user_id.strip()}::{purpose}"


def set_owner_supplied_secret(user_id: str, purpose: str, value: str) -> None:
    """Record the key the owner obtained for this specific subscriber."""
    if not value or not value.strip():
        raise ValueError("refusing to store an empty credential")
    from backend.services.usb_vault import set_secret
    set_secret(owner_supplied_vault_key(user_id, purpose), value.strip())


def get_owner_supplied_secret(user_id: str, purpose: str) -> str | None:
    from backend.services.usb_vault import get_secret
    try:
        v = get_secret(owner_supplied_vault_key(user_id, purpose))
    except Exception as exc:
        log.warning("vault read failed for %s/%s: %s", user_id, purpose, exc)
        return None
    return v.strip() if v and v.strip() else None


def missing_owner_supplied(user_id: str) -> list[str]:
    """Which owner-supplied credentials this subscriber does NOT yet have.

    This is the list provisioning must refuse to proceed past, because none of them can
    be generated — only the owner can obtain them.
    """
    return [p for p in OWNER_SUPPLIED_PURPOSES if not get_owner_supplied_secret(user_id, p)]


def all_subscriber_credentials(user_id: str, version: int = 1) -> dict[str, Any]:
    """The complete credential picture for one subscriber: derived AND owner-supplied.

    Deliberately reports the two categories separately, and names what is missing, rather
    than merging them into one dict where an absent owner-supplied key would look
    identical to a present one.
    """
    derived = subscriber_secrets(user_id, version)
    supplied = {p: get_owner_supplied_secret(user_id, p) for p in OWNER_SUPPLIED_PURPOSES}
    missing = [p for p, v in supplied.items() if not v]
    return {
        "user_id": user_id,
        "derived": derived,
        "owner_supplied_present": sorted(p for p, v in supplied.items() if v),
        "owner_supplied_missing": missing,
        "complete": not missing,
        "action_required_by_owner": [
            {"purpose": p,
             "what": OWNER_SUPPLIED_PURPOSES[p][0],
             "how": OWNER_SUPPLIED_PURPOSES[p][1]}
            for p in missing
        ],
    }


def space_id_for(user_id: str, owner_namespace: str) -> str:
    """Stable, collision-free Space id for a subscriber.

    The user id is hashed rather than embedded, so a subscriber's account identifier
    (which may be an email or another personal identifier) never becomes part of a
    publicly visible Space name.
    """
    digest = hashlib.sha256(_encode("space-id", user_id.strip())).hexdigest()[:16]
    return f"{owner_namespace}/omega-sub-{digest}"


def verify_no_collisions(user_ids: list[str], version: int = 1) -> dict[str, Any]:
    """Prove uniqueness rather than assume it (credentials-and-secrets.md:45).

    Checks that across the given users: no two share any derived secret, no two share a
    Space id, and no single user reuses one value across two different purposes.
    """
    seen_secret: dict[str, tuple[str, str]] = {}
    seen_space: dict[str, str] = {}
    collisions: list[str] = []
    for uid in user_ids:
        sid = space_id_for(uid, "acme")
        if sid in seen_space and seen_space[sid] != uid:
            collisions.append(f"space id collision: {uid} vs {seen_space[sid]}")
        seen_space[sid] = uid
        for purpose, value in subscriber_secrets(uid, version).items():
            if value in seen_secret:
                prev_uid, prev_purpose = seen_secret[value]
                collisions.append(
                    f"secret collision: ({uid},{purpose}) == ({prev_uid},{prev_purpose})"
                )
            seen_secret[value] = (uid, purpose)
    return {
        "users": len(user_ids),
        "distinct_secrets": len(seen_secret),
        "expected_secrets": len(user_ids) * len(SUBSCRIBER_SECRET_PURPOSES),
        "distinct_space_ids": len(seen_space),
        "collisions": collisions,
        "ok": not collisions
        and len(seen_secret) == len(user_ids) * len(SUBSCRIBER_SECRET_PURPOSES)
        and len(seen_space) == len(user_ids),
    }


def rotate_only_this_user(user_id: str, current_version: int) -> dict[str, str]:
    """Rotation writes to exactly one subscriber's entries and nothing else.

    credentials-and-secrets.md:45 asks for a real end-to-end rotation trace confirming it
    updates the intended user's credential and no other. Bumping the version changes
    every value for this user and, by construction, cannot alter any other user's, since
    user_id is part of every derivation input.
    """
    return subscriber_secrets(user_id, current_version + 1)


@dataclass
class ProvisionResult:
    user_id: str
    space_id: str
    secrets_set: list[str] = field(default_factory=list)
    created: bool = False
    dry_run: bool = True
    skipped_reason: str | None = None
    errors: list[str] = field(default_factory=list)
    # v22-2: owner-supplied credentials this subscriber is still missing. Non-empty means
    # a human step is outstanding — the Space can be created, but the connectors those
    # keys serve will not work until the owner supplies them.
    owner_supplied_missing: list[str] = field(default_factory=list)
    owner_action_required: list[dict[str, str]] = field(default_factory=list)


def provision_subscriber_space(
    user_id: str,
    *,
    owner_namespace: str,
    template_space: str,
    hf_token: str | None = None,
    version: int = 1,
    dry_run: bool = True,
) -> ProvisionResult:
    """Create (or re-configure) one subscriber's dedicated free-tier Space.

    dry_run=True (the default) performs every step except the outward-facing HF calls,
    so the whole path — derivation, naming, the exact secret set — can be exercised and
    asserted without creating anything real.

    The owner is never provisioned a per-subscriber Space
    (huggingface-spaces-hosting.md:51).
    """
    space_id = space_id_for(user_id, owner_namespace)
    result = ProvisionResult(user_id=user_id, space_id=space_id, dry_run=dry_run)

    if user_id.strip() == OWNER_USER_ID:
        result.skipped_reason = (
            "owner plan keeps its own dedicated hosting; per-subscriber provisioning "
            "does not apply (huggingface-spaces-hosting.md:51)"
        )
        return result

    values = subscriber_secrets(user_id, version)
    result.secrets_set = sorted(values)

    # v22-2: report owner-supplied credentials BEFORE the dry-run return, so a dry run —
    # which is how this path is normally exercised — surfaces the outstanding human step
    # rather than reporting a clean provision that would be incomplete in reality.
    credentials = all_subscriber_credentials(user_id, version)
    result.owner_supplied_missing = credentials["owner_supplied_missing"]
    result.owner_action_required = credentials["action_required_by_owner"]
    for purpose in result.owner_supplied_missing:
        what, how = OWNER_SUPPLIED_PURPOSES[purpose]
        log.warning(
            "subscriber %s has no %s. It cannot be derived — %s Until then the features "
            "it serves will fail for this subscriber.", user_id, purpose, how,
        )
    # Owner-supplied values are injected alongside the derived ones where present, so a
    # subscriber who HAS been given a key gets it in their own Space's secrets.
    for purpose in OWNER_SUPPLIED_PURPOSES:
        supplied = get_owner_supplied_secret(user_id, purpose)
        if supplied:
            values[purpose] = supplied
            result.secrets_set = sorted(values)

    if dry_run:
        return result

    token = hf_token or os.environ.get("HF_API_TOKEN") or os.environ.get("HF_TOKEN") or ""
    if not token:
        result.errors.append("no HF_API_TOKEN/HF_TOKEN available")
        return result

    try:
        from huggingface_hub import HfApi
        api = HfApi(token=token)
        try:
            api.space_info(space_id)
            log.info("space %s already exists; reconfiguring secrets only", space_id)
        except Exception:
            api.duplicate_space(from_id=template_space, to_id=space_id, private=True)
            result.created = True
        for purpose, value in values.items():
            api.add_space_secret(repo_id=space_id, key=purpose, value=value)
        api.restart_space(repo_id=space_id)
    except Exception as exc:  # pragma: no cover - network path
        result.errors.append(f"{type(exc).__name__}: {exc}")
    return result


def new_root_secret() -> str:
    """Generate a suitable SUBSCRIBER_ROOT_SECRET (operator runs this once)."""
    return secrets.token_urlsafe(48)