pigway-frame-flow/tests/test_accounts.py

import hashlib
import time

from pigway_frame_flow.core.accounts import password_hash


def account(app):
    app.state.service.db.execute(
        "INSERT INTO users VALUES(?,?,1)",
        ("test-user", password_hash("fixture-password")),
    )


def login(client, app, remember=True):
    account(app)
    client.headers.pop("Authorization", None)
    return client.post(
        "/api/v1/auth/login",
        json={
            "username": "test-user",
            "password": "fixture-password",
            "remember": remember,
        },
    )


def test_session_login_and_csrf(client, app):
    response = login(client, app)
    assert response.status_code == 200
    assert "HttpOnly" in response.headers["set-cookie"]
    assert "SameSite=strict" in response.headers["set-cookie"]
    assert client.get("/api/v1/devices").status_code == 200
    assert client.post("/api/v1/jobs", json={}).status_code == 403
    assert client.get("/api/v1/auth/me").json()["username"] == "test-user"
    client.headers["X-CSRF-Token"] = response.json()["csrf_token"]
    assert client.post("/api/v1/auth/logout").status_code == 204
    assert client.get("/api/v1/devices").status_code == 401


def test_password_change_revokes_sessions(client, app):
    response = login(client, app)
    client.headers["X-CSRF-Token"] = response.json()["csrf_token"]
    assert (
        client.put(
            "/api/v1/auth/password",
            json={"current_password": "wrong", "new_password": "replacement-password"},
        ).status_code
        == 401
    )
    assert (
        client.put(
            "/api/v1/auth/password",
            json={
                "current_password": "fixture-password",
                "new_password": "replacement-password",
            },
        ).status_code
        == 204
    )
    assert client.get("/api/v1/auth/me").status_code == 401
    assert (
        client.post(
            "/api/v1/auth/login",
            json={"username": "test-user", "password": "replacement-password"},
        ).status_code
        == 200
    )


def test_login_rate_limit(client, app):
    account(app)
    for _ in range(5):
        assert (
            client.post(
                "/api/v1/auth/login",
                json={"username": "test-user", "password": "wrong"},
            ).status_code
            == 401
        )
    assert (
        client.post(
            "/api/v1/auth/login",
            json={"username": "test-user", "password": "fixture-password"},
        ).status_code
        == 429
    )


def test_session_expiry_and_no_plaintext(client, app):
    response = login(client, app, False)
    assert "Max-Age" not in response.headers["set-cookie"]
    raw = client.cookies.get("frameflow_session")
    row = app.state.service.db.one("SELECT * FROM sessions")
    assert row["token_hash"] == hashlib.sha256(raw.encode()).hexdigest()
    app.state.service.db.execute("UPDATE sessions SET expires_at=?", (time.time() - 1,))
    assert client.get("/api/v1/auth/me").status_code == 401
    assert "fixture-password" not in str(
        app.state.service.db.one("SELECT * FROM users")
    )


def test_reset_template(client):
    assert (
        client.put(
            "/api/v1/templates/quote/config", json={"data": {"title": "Saved title"}}
        ).status_code
        == 200
    )
    assert (
        client.get("/api/v1/templates/quote").json()["defaults"]["title"]
        == "Saved title"
    )
    assert client.delete("/api/v1/templates/quote/config").status_code == 204
    assert (
        client.get("/api/v1/templates/quote").json()["defaults"]["title"]
        != "Saved title"
    )


def test_openapi_describes_cookie_auth_and_docs_csrf(client):
    schema = client.get("/openapi.json").json()
    cookie = schema["components"]["securitySchemes"]["AccountSession"]
    assert cookie["in"] == "cookie" and cookie["name"] == "frameflow_session"
    assert {"AccountSession": []} in schema["paths"]["/api/v1/devices"]["get"][
        "security"
    ]
    assert "X-CSRF-Token" in client.get("/docs").text


def test_browser_admin_can_read_other_clients_operation(client, app, service):
    operation_id = service.submit(
        "render",
        {"template": "quote"},
        "living-room",
        "scheduler",
        "fixture-request",
        None,
    )
    assert login(client, app).status_code == 200
    assert client.get("/api/v1/operations/" + operation_id).status_code == 200
    assert operation_id in [
        op["operation_id"] for op in client.get("/api/v1/operations").json()
    ]


def test_password_minimum_and_https_cookie(client, app):
    response = login(client, app)
    client.headers["X-CSRF-Token"] = response.json()["csrf_token"]
    assert (
        client.put(
            "/api/v1/auth/password",
            json={"current_password": "fixture-password", "new_password": "1234567"},
        ).status_code
        == 422
    )
    assert (
        client.put(
            "/api/v1/auth/password",
            json={"current_password": "fixture-password", "new_password": "12345678"},
        ).status_code
        == 204
    )
    response = client.post(
        "https://testserver/api/v1/auth/login",
        json={"username": "test-user", "password": "12345678"},
    )
    assert response.status_code == 200
    assert "Secure" in response.headers["set-cookie"]
    assert response.headers["cache-control"] == "no-store"


def test_login_rejects_cross_origin(client, app):
    account(app)
    assert (
        client.post(
            "/api/v1/auth/login",
            headers={"Origin": "https://evil.example"},
            json={"username": "test-user", "password": "fixture-password"},
        ).status_code
        == 403
    )


def test_account_throttle_across_source_addresses(client, app):
    account(app)
    for _ in range(5):
        assert (
            client.post(
                "/api/v1/auth/login",
                json={"username": "test-user", "password": "wrong"},
            ).status_code
            == 401
        )
        app.state.service.db.execute(
            "DELETE FROM login_attempts WHERE identity NOT LIKE 'account:%'"
        )
    assert (
        client.post(
            "/api/v1/auth/login",
            json={"username": "test-user", "password": "fixture-password"},
        ).status_code
        == 429
    )


def test_legacy_hash_upgrade(client, app):
    import hmac

    from pigway_frame_flow.core.accounts import verify_password

    salt = "12" * 16
    digest = hashlib.scrypt(
        b"fixture-password", salt=bytes.fromhex(salt), n=16384, r=8, p=1
    ).hex()
    legacy = f"scrypt${salt}${digest}"
    app.state.service.db.execute(
        "INSERT INTO users VALUES(?,?,1)", ("test-user", legacy)
    )
    assert verify_password("fixture-password", legacy)
    assert (
        client.post(
            "/api/v1/auth/login",
            json={"username": "test-user", "password": "fixture-password"},
        ).status_code
        == 200
    )
    stored = app.state.service.db.one("SELECT password_hash FROM users")[
        "password_hash"
    ]
    assert stored.startswith("scrypt-v2$") and not hmac.compare_digest(stored, legacy)
    assert verify_password("fixture-password", stored)


def test_only_one_browser_account(app):
    import sqlite3

    import pytest

    account(app)
    with pytest.raises(sqlite3.IntegrityError):
        app.state.service.db.execute(
            "INSERT INTO users VALUES(?,?,1)",
            ("other", password_hash("another-password")),
        )