cista-storage/tests/test_tokens.py

from pathlib import Path
from time import time
from uuid import uuid4

import os

import pytest
import pytest_asyncio
from sanic import Sanic

from cista import auth, config, watching
from cista.auth import bp as auth_bp


def _persist_config():
    import msgspec
    from pathlib import PurePath

    def enc_hook(obj):
        if isinstance(obj, PurePath):
            return obj.as_posix()
        raise TypeError

    raw = msgspec.to_builtins(config.config, enc_hook=enc_hook)
    config.conffile.write_bytes(msgspec.toml.encode(raw))


@pytest.fixture()
def setup_storage(tmp_path: Path):
    os.environ["CISTA_HOME"] = str(tmp_path)
    config.init_confdir()
    user = config.User()
    auth.set_password(user, "secret")
    admin = config.User(privileged=True)
    auth.set_password(admin, "secret")
    config.config = config.Config(
        path=tmp_path,
        listen=":0",
        public=False,
        users={"alice": user, "admin": admin},
    )
    _persist_config()
    watching.state.root = []
    watching.rootpath = tmp_path
    yield tmp_path
    watching.state.root = []


@pytest_asyncio.fixture()
async def client(setup_storage: Path):
    app = Sanic(f"token-test-{uuid4().hex}", strict_slashes=True)
    app.router.ALLOWED_METHODS = (
        *app.router.ALLOWED_METHODS,
        "MKCOL",
        "MOVE",
        "COPY",
        "PROPFIND",
    )
    app.blueprint(auth_bp)
    yield app.asgi_client


def _basic_auth(username: str, password: str) -> str:
    return f"Basic {__import__('base64').b64encode(f'{username}:{password}'.encode()).decode()}"


@pytest.mark.asyncio
async def test_token_crud(client):
    # Admin creates a token without specifying username (auto-assigned)
    _, res = await client.post(
        "/auth/tokens",
        json={"name": "test"},
        headers={"Authorization": _basic_auth("admin", "secret")},
    )
    assert res.status_code == 200
    data = res.json
    assert "id" in data
    assert "key" in data
    assert data["username"] == "admin"
    assert data["name"] == "test"
    token_id = data["id"]
    token_key = data["key"]

    # List tokens - admin sees only their own
    _, res = await client.get(
        "/auth/tokens",
        headers={"Authorization": _basic_auth("admin", "secret")},
    )
    assert res.status_code == 200
    tokens = res.json["tokens"]
    assert len(tokens) == 1
    assert tokens[0]["id"] == token_id
    assert tokens[0]["username"] == "admin"

    # Use token via Basic auth (token:<secret>)
    _, res = await client.get(
        "/auth/tokens",
        headers={"Authorization": _basic_auth("token", token_key)},
    )
    assert res.status_code == 200

    # Delete token
    _, res = await client.delete(
        f"/auth/tokens/{token_id}",
        headers={"Authorization": _basic_auth("admin", "secret")},
    )
    assert res.status_code == 200

    # List should be empty
    _, res = await client.get(
        "/auth/tokens",
        headers={"Authorization": _basic_auth("admin", "secret")},
    )
    assert res.status_code == 200
    assert len(res.json["tokens"]) == 0


@pytest.mark.asyncio
async def test_token_user_scoped(client):
    # Alice creates a token for herself (no username specified)
    _, res = await client.post(
        "/auth/tokens",
        json={"name": "alice-token"},
        headers={"Authorization": _basic_auth("alice", "secret")},
    )
    assert res.status_code == 200
    alice_token_id = res.json["id"]
    alice_token_key = res.json["key"]

    # Admin creates a token for themselves
    _, res = await client.post(
        "/auth/tokens",
        json={"name": "admin-token"},
        headers={"Authorization": _basic_auth("admin", "secret")},
    )
    assert res.status_code == 200
    admin_token_id = res.json["id"]

    # Alice lists tokens - sees only her own
    _, res = await client.get(
        "/auth/tokens",
        headers={"Authorization": _basic_auth("alice", "secret")},
    )
    assert res.status_code == 200
    tokens = res.json["tokens"]
    assert len(tokens) == 1
    assert tokens[0]["id"] == alice_token_id
    assert tokens[0]["username"] == "alice"

    # Admin lists tokens - sees only their own
    _, res = await client.get(
        "/auth/tokens",
        headers={"Authorization": _basic_auth("admin", "secret")},
    )
    assert res.status_code == 200
    tokens = res.json["tokens"]
    assert len(tokens) == 1
    assert tokens[0]["id"] == admin_token_id
    assert tokens[0]["username"] == "admin"

    # Alice cannot create a token for admin
    _, res = await client.post(
        "/auth/tokens",
        json={"username": "admin", "name": "impersonation"},
        headers={"Authorization": _basic_auth("alice", "secret")},
    )
    assert res.status_code == 403

    # Alice cannot delete admin's token
    _, res = await client.delete(
        f"/auth/tokens/{admin_token_id}",
        headers={"Authorization": _basic_auth("alice", "secret")},
    )
    assert res.status_code == 403

    # Alice can delete her own token
    _, res = await client.delete(
        f"/auth/tokens/{alice_token_id}",
        headers={"Authorization": _basic_auth("alice", "secret")},
    )
    assert res.status_code == 200

    # Alice's token auth still works until deletion is processed
    # Verify token auth worked during the test
    _, res = await client.get(
        "/auth/tokens",
        headers={"Authorization": _basic_auth("token", alice_token_key)},
    )
    # Token was deleted above, so this should now be unauthenticated
    # Actually the token key lookup will fail, and since there's no session fallback...
    # With auth header present but invalid, it should return 401
    assert res.status_code == 401