ounce100m_credentials.py

"""Runtime credentials for ounce100m jobs, fetched from the account's own private Kaggle dataset.

Why this exists. §2 forbids the HF token reaching a public artifact -- job logs, model cards, or any
code mirrored to the Hub -- and requires it to come "from the environment or a secret store". Internet-
enabled Kaggle sessions arrive already authenticated against the Kaggle API (verified: CLI 2.0.2
preinstalled, `auth_method: ACCESS_TOKEN`), and `dodosoomro/ounce100m-secret-store` is a private dataset
that is anonymously unreachable and absent from public search. So a job can retrieve its own credential
at run time: no token in kernel source, no token in the public code repo, no token in any log.

Note that `datasetDataSources` mounting was tried and did NOT populate /kaggle/input (see
memory/ERRORS.md), so this module downloads instead of mounting. Download also has the advantage of
working in a freshly created kernel with no metadata to keep in sync.

Rules for callers:
  * never print, log, or write the token to /kaggle/working -- working-dir files become kernel outputs
  * call `install()` once at process start, before any huggingface_hub import
  * the staging directory is under /tmp, which is not published as an artifact
"""

import json
import os
import subprocess
import zipfile

DS_SLUG = "ounce100m-secret-store"
DS_ID = f"dodosoomro/{DS_SLUG}"
STAGE = "/tmp/.ounce100m"  # deliberately not under /kaggle/working
CREDS = "credentials.json"


class CredentialError(RuntimeError):
    pass


def _download():
    os.makedirs(STAGE, exist_ok=True)
    os.chmod(STAGE, 0o700)
    r = subprocess.run(
        ["kaggle", "datasets", "download", "-d", DS_ID, "-p", STAGE, "--unzip"],
        capture_output=True, text=True, timeout=300)
    if r.returncode != 0:
        # Scrubbed: an error string from the CLI could otherwise echo the token into a retained log.
        raise CredentialError(f"kaggle datasets download rc={r.returncode}")
    path = os.path.join(STAGE, CREDS)
    if not os.path.exists(path):
        # --unzip flattens differently across CLI versions; fall back to opening the archive directly.
        zipped = os.path.join(STAGE, f"{DS_SLUG}.zip")
        if os.path.exists(zipped):
            with zipfile.ZipFile(zipped) as z:
                z.extract(CREDS, STAGE)
            path = os.path.join(STAGE, CREDS)
    if not os.path.exists(path):
        raise CredentialError(f"{CREDS} not found after download; staged: {sorted(os.listdir(STAGE))}")
    return path


def token():
    """The HF token as a string. Callers must not print it."""
    path = os.environ.get("OUNCE100M_CRED_PATH")  # lets a mounting job hand us a path instead
    if not path:
        path = _download()
    with open(path) as f:
        blob = json.load(f)
    tok = blob.get("HF_TOKEN", "")
    if not tok.startswith("hf_") or len(tok) < 20:
        raise CredentialError("credential file present but HF_TOKEN malformed -- refusing to continue")
    return tok


def install(verify=False):
    """Put the token where huggingface_hub expects it. Returns a *safe to print* summary only."""
    import hashlib

    tok = token()
    os.environ["HF_TOKEN"] = tok
    os.environ["HUGGING_FACE_HUB_TOKEN"] = tok
    out = {"source": DS_ID, "length": len(tok),
           "sha256_prefix": hashlib.sha256(tok.encode()).hexdigest()[:12]}
    if verify:
        from huggingface_hub import HfApi

        me = HfApi(token=tok).whoami()
        out["hub_user"] = me.get("name")
    return out


if __name__ == "__main__":
    print(json.dumps(install(verify=True), indent=1))