stackpilot/backend/tests/test_registries.py

"""Private registry credentials.

Three things have to hold, and the first one is a bug fix rather than a feature:

* a private image whose registry refuses us must *say so*. Before this, a 401
  was indistinguishable from a network blip, so the UI showed nothing and a
  stack sitting on a months-old image looked up to date.
* the host people type has to reach the host an image reference parses to.
  Docker Hub has five spellings and `parse_ref` only ever produces one of them,
  so credentials entered as "docker.io" would otherwise never be found.
* the password must be encrypted at rest and must never come back out of the
  API — not even masked, because the UI has no use for it.
"""
from __future__ import annotations

import base64
import json
import os
import stat

import pytest


@pytest.fixture
def svc(db):
    from services import registry_service

    return registry_service


@pytest.fixture(autouse=True)
def clean_registries(db):
    """Credentials are global state; don't let one test leak into the next."""
    from sqlmodel import Session, delete

    from database import engine
    from models.registry import Registry
    from services import registry_service

    def wipe():
        with Session(engine) as session:
            session.exec(delete(Registry))
            session.commit()
            registry_service.reload(session)

    wipe()
    yield
    wipe()


def _add(host: str, username: str = "bob", password: str = "hunter2"):
    """Insert a registry the way the API would, and refresh the cache."""
    from sqlmodel import Session

    from database import engine
    from models.registry import Registry
    from services import crypto_service, registry_service

    with Session(engine) as session:
        session.add(
            Registry(
                name=host,
                host=registry_service.canonical_host(host),
                username=username,
                password=crypto_service.encrypt(password),
            )
        )
        session.commit()
        registry_service.reload(session)


# --------------------------------------------------------------------------- #
# Host normalization
# --------------------------------------------------------------------------- #


@pytest.mark.parametrize(
    "written",
    [
        "docker.io",
        "index.docker.io",
        "registry-1.docker.io",
        "https://index.docker.io/v1/",
        "https://docker.io",
        "DOCKER.IO",
        "  docker.io/  ",
    ],
)
def test_every_spelling_of_docker_hub_lands_on_one_host(svc, written):
    """`parse_ref` only ever says registry-1.docker.io; the lookup has to agree."""
    assert svc.canonical_host(written) == svc.DOCKER_HUB


@pytest.mark.parametrize(
    "written,expected",
    [
        ("ghcr.io", "ghcr.io"),
        ("https://ghcr.io/", "ghcr.io"),
        ("registry.example.com:5000", "registry.example.com:5000"),
        ("http://registry.lan:5000/v2/", "registry.lan:5000"),
        ("GHCR.IO", "ghcr.io"),
    ],
)
def test_other_registries_keep_their_host_and_port(svc, written, expected):
    assert svc.canonical_host(written) == expected


def test_an_empty_host_is_refused(svc):
    for value in ("", "   ", "https://"):
        with pytest.raises(svc.RegistryError):
            svc.canonical_host(value)


# --------------------------------------------------------------------------- #
# The lookup the update checker uses
# --------------------------------------------------------------------------- #


def test_credentials_are_found_for_the_registry_an_image_names(svc):
    _add("ghcr.io", "bob", "hunter2")
    assert svc.credentials_for("ghcr.io") == ("bob", "hunter2")
    assert svc.credentials_for_image("ghcr.io/acme/app:1.2") == ("bob", "hunter2")


def test_docker_hub_credentials_are_found_for_a_bare_image_name(svc):
    # "nginx:alpine" parses to registry-1.docker.io; the row was typed as
    # "docker.io". This is the join that makes the whole feature work.
    _add("docker.io", "bob", "hunter2")
    assert svc.credentials_for_image("nginx:alpine") == ("bob", "hunter2")
    assert svc.credentials_for_image("acme/private:1.0") == ("bob", "hunter2")


def test_an_unconfigured_registry_has_no_credentials(svc):
    _add("ghcr.io")
    assert svc.credentials_for("quay.io") is None
    assert svc.credentials_for_image("quay.io/acme/app") is None


def test_a_row_that_cannot_be_decrypted_does_not_break_the_others(svc, monkeypatch):
    """A changed SECRET_KEY must cost you one registry, not all of them."""
    from sqlmodel import Session

    from database import engine
    from models.registry import Registry

    _add("ghcr.io", "bob", "hunter2")
    with Session(engine) as session:
        session.add(Registry(name="broken", host="quay.io", username="x", password="enc:v1:nonsense"))
        session.commit()
        svc.reload(session)
    assert svc.credentials_for("ghcr.io") == ("bob", "hunter2")
    assert svc.credentials_for("quay.io") is None


# --------------------------------------------------------------------------- #
# The file the Docker CLI reads
# --------------------------------------------------------------------------- #


def _config(svc) -> dict:
    with open(os.path.join(svc.docker_config_dir(), "config.json"), encoding="utf-8") as fh:
        return json.load(fh)


def test_the_cli_config_carries_a_usable_auth_entry(svc):
    _add("ghcr.io", "bob", "hunter2")
    auths = _config(svc)["auths"]
    token = auths["ghcr.io"]["auth"]
    assert base64.b64decode(token).decode() == "bob:hunter2"


def test_docker_hub_is_written_under_the_key_the_cli_expects(svc):
    # The CLI looks Docker Hub up as https://index.docker.io/v1/, not as the
    # host an image reference parses to.
    _add("docker.io", "bob", "hunter2")
    assert svc.DOCKER_HUB_CONFIG_KEY in _config(svc)["auths"]


def test_the_cli_config_is_not_readable_by_anyone_else(svc):
    _add("ghcr.io")
    path = os.path.join(svc.docker_config_dir(), "config.json")
    assert stat.S_IMODE(os.stat(path).st_mode) == 0o600


def test_removing_a_registry_revokes_the_cli_login_too(svc):
    _add("ghcr.io")
    assert "ghcr.io" in _config(svc)["auths"]
    # clean_registries' wipe() is exactly what deleting the last row does.
    from sqlmodel import Session, delete

    from database import engine
    from models.registry import Registry

    with Session(engine) as session:
        session.exec(delete(Registry))
        session.commit()
        svc.reload(session)
    assert _config(svc)["auths"] == {}


def test_compose_runs_against_our_config_not_the_home_directory(svc):
    assert svc.cli_env()["DOCKER_CONFIG"] == svc.docker_config_dir()


# --------------------------------------------------------------------------- #
# Through the API
# --------------------------------------------------------------------------- #


def test_create_list_and_delete(as_admin):
    created = as_admin.post(
        "/api/registries",
        json={"name": "GHCR", "host": "https://ghcr.io/", "username": "bob", "password": "hunter2"},
    )
    assert created.status_code == 201, created.text
    body = created.json()
    assert body["host"] == "ghcr.io"  # normalized on the way in
    assert body["has_password"] is True

    rows = as_admin.get("/api/registries").json()
    assert [r["host"] for r in rows] == ["ghcr.io"]

    assert as_admin.delete(f"/api/registries/{body['id']}").status_code == 200
    assert as_admin.get("/api/registries").json() == []


def test_the_password_never_comes_back_out(as_admin):
    as_admin.post(
        "/api/registries",
        json={"host": "ghcr.io", "username": "bob", "password": "hunter2"},
    )
    payload = as_admin.get("/api/registries").text
    assert "hunter2" not in payload
    # No password field at all — only the has_password flag the UI needs to
    # decide whether to show the input as "leave blank to keep".
    assert '"password"' not in payload
    assert '"has_password":true' in payload.replace(" ", "")


def test_the_password_is_encrypted_at_rest(as_admin):
    from sqlmodel import Session, select

    from database import engine
    from models.registry import Registry
    from services import crypto_service

    as_admin.post(
        "/api/registries",
        json={"host": "ghcr.io", "username": "bob", "password": "hunter2"},
    )
    with Session(engine) as session:
        row = session.exec(select(Registry)).one()
    assert "hunter2" not in row.password
    assert crypto_service.is_encrypted(row.password)
    assert crypto_service.decrypt(row.password) == "hunter2"


def test_two_rows_for_the_same_registry_are_refused(as_admin):
    first = {"host": "ghcr.io", "username": "bob", "password": "x"}
    assert as_admin.post("/api/registries", json=first).status_code == 201
    # The same registry under a different spelling is still the same registry.
    second = {"host": "https://ghcr.io", "username": "alice", "password": "y"}
    assert as_admin.post("/api/registries", json=second).status_code == 409


def test_editing_without_a_password_keeps_the_stored_one(as_admin, svc):
    created = as_admin.post(
        "/api/registries",
        json={"host": "ghcr.io", "username": "bob", "password": "hunter2"},
    ).json()
    updated = as_admin.put(
        f"/api/registries/{created['id']}", json={"username": "alice"}
    )
    assert updated.status_code == 200, updated.text
    # The UI never received the password, so it cannot send it back — and must
    # not have to.
    assert svc.credentials_for("ghcr.io") == ("alice", "hunter2")


def test_a_saved_password_is_replaced_when_one_is_given(as_admin, svc):
    created = as_admin.post(
        "/api/registries",
        json={"host": "ghcr.io", "username": "bob", "password": "hunter2"},
    ).json()
    as_admin.put(f"/api/registries/{created['id']}", json={"password": "correcthorse"})
    assert svc.credentials_for("ghcr.io") == ("bob", "correcthorse")


def test_a_nonsense_host_is_a_400(as_admin):
    response = as_admin.post(
        "/api/registries", json={"host": "   ", "username": "b", "password": "p"}
    )
    assert response.status_code == 400


def test_the_read_only_role_cannot_see_or_touch_registries(as_user):
    assert as_user.get("/api/registries").status_code == 403
    assert (
        as_user.post(
            "/api/registries", json={"host": "ghcr.io", "username": "b", "password": "p"}
        ).status_code
        == 403
    )


# --------------------------------------------------------------------------- #
# The bug this was really about
# --------------------------------------------------------------------------- #


def _refusing(monkeypatch):
    """Point check_image at a registry that answers 401."""
    from services import update_service

    async def refuse(image: str):
        raise update_service.AuthRequired("ghcr.io")

    monkeypatch.setattr(update_service, "remote_digest", refuse)
    monkeypatch.setattr(update_service, "_local_digest", lambda image: "sha256:local")
    return update_service


def test_a_registry_that_refuses_us_is_reported_not_swallowed(monkeypatch):
    """The whole point: 401 must not look like "up to date"."""
    import asyncio

    update_service = _refusing(monkeypatch)
    result = asyncio.run(update_service.check_image("ghcr.io/acme/private:1.0"))

    assert result.update_available is False
    # Names the registry, because the fix is "add credentials for that one".
    assert "ghcr.io" in result.error
    assert "needs credentials" in result.error


def test_stored_credentials_that_are_rejected_say_so(monkeypatch):
    import asyncio

    _add("ghcr.io", "bob", "hunter2")
    update_service = _refusing(monkeypatch)
    result = asyncio.run(update_service.check_image("ghcr.io/acme/private:1.0"))

    # Different wording, because the fix is different: the credentials are there
    # and wrong, rather than missing.
    assert "rejected" in result.error


def test_a_registry_we_simply_cannot_reach_still_says_that(monkeypatch):
    """The old message has to survive — not every failure is an auth failure."""
    import asyncio

    from services import update_service

    async def unreachable(image: str):
        return None

    monkeypatch.setattr(update_service, "remote_digest", unreachable)
    monkeypatch.setattr(update_service, "_local_digest", lambda image: "sha256:local")
    result = asyncio.run(update_service.check_image("ghcr.io/acme/app:1.0"))
    assert result.error == "could not reach registry"