PacketStation/tests/server/test_signatures.py

"""signatures.sign: step-up, the append-only row, its content hash and the audit row."""

from __future__ import annotations

import hashlib
import uuid
from typing import Any

import psycopg
import pytest
from fakedir import FAKE_PASSWORD, FakeDirectory
from fastapi.testclient import TestClient
from pshelpers import PIN, app_conn, needs_pg, set_pin, sign_in

from packetstation import audit
from packetstation.api.errors import ApiError
from packetstation.auth import sessions, signatures
from packetstation.settings import Settings


def test_content_hash_is_canonical() -> None:
    a = signatures.content_hash({"b": [1, 2], "a": "7A"})
    b = signatures.content_hash({"a": "7A", "b": [1, 2]})
    assert a == b == hashlib.sha256(b'{"a":"7A","b":[1,2]}').digest()
    assert signatures.content_hash(None) is None


def _actor(conn: Any, sam: str) -> sessions.Actor:
    row = conn.execute(
        "SELECT s.id, u.id, u.sam_account, u.display_name, s.kind::text"
        " FROM sessions s JOIN users u ON u.id = s.user_id"
        " WHERE u.sam_account = %s AND s.revoked_at IS NULL ORDER BY s.created_at DESC LIMIT 1",
        (sam,),
    ).fetchone()
    assert row is not None
    return sessions.Actor(row[0], row[1], row[2], row[3], row[4], ("shop:03:worker",))


@needs_pg
def test_sign_with_pin_and_password(pg_client: TestClient, pg_settings: Settings, fake_dir: FakeDirectory) -> None:
    set_pin(pg_client, "weld.andy")
    sign_in(pg_client, "weld.andy")
    content = {"packet": "L662001/1", "lot": "7A", "step": "submit"}
    with app_conn(pg_settings) as conn:
        actor = _actor(conn, "weld.andy")
        sig = signatures.sign(
            conn,
            fake_dir,
            actor,
            factor="pin",
            secret=PIN,
            action="packet.submit",
            meaning="Work complete",
            packet_id=12,
            subject_type="packet",
            subject_id=12,
            content=content,
            settings=pg_settings,
        )
        assert sig.content_hash == signatures.content_hash(content) and sig.factor == "pin"
        sig2 = signatures.sign(
            conn,
            fake_dir,
            actor,
            factor="password",
            secret=FAKE_PASSWORD,
            action="x",
            meaning=None,
            settings=pg_settings,
        )
        assert sig2.id > sig.id and sig2.content_hash is None
        row = conn.execute(
            "SELECT account, display_name, session_kind::text, factor::text, action, packet_id, subject_id,"
            " content_hash FROM signatures WHERE id = %s",
            (sig.id,),
        ).fetchone()
        assert row[:7] == ("weld.andy", "Andy Welder", "password", "pin", "packet.submit", 12, "12")
        assert bytes(row[7]) == signatures.content_hash(content)
        detail = conn.execute("SELECT detail FROM audit_events WHERE action = 'sig.signed' ORDER BY id").fetchone()[0]
        assert detail == {"signature_id": sig.id, "action": "packet.submit", "factor": "pin"}
        for stmt in ("UPDATE signatures SET action = 'x'", "DELETE FROM signatures", "TRUNCATE signatures"):
            with pytest.raises(psycopg.errors.InsufficientPrivilege):
                conn.execute(stmt)
        for stmt in ("UPDATE signin_attempts SET ok = true", "DELETE FROM signin_attempts"):
            with pytest.raises(psycopg.errors.InsufficientPrivilege):
                conn.execute(stmt)
        assert audit.verify_chain(conn)[0]


@needs_pg
def test_wrong_signatures_end_the_session(
    pg_client: TestClient, pg_settings: Settings, fake_dir: FakeDirectory
) -> None:
    sign_in(pg_client, "weld.andy")
    with app_conn(pg_settings) as conn:
        actor = _actor(conn, "weld.andy")
        for code in ("wrong_secret", "wrong_secret", "session_ended"):
            with pytest.raises(ApiError) as e:
                signatures.sign(
                    conn,
                    fake_dir,
                    actor,
                    factor="password",
                    secret="nope",
                    action="x",
                    meaning=None,
                    settings=pg_settings,
                )
            assert e.value.code == code
        assert conn.execute("SELECT count(*) FROM signatures").fetchone()[0] == 0
        kinds = conn.execute("SELECT DISTINCT kind FROM signin_attempts WHERE NOT ok").fetchall()
        assert kinds == [("signature",)]
    assert pg_client.get("/api/auth/me").json()["code"] == "no_session"


def test_unknown_factor_refused() -> None:
    actor = sessions.Actor(uuid.uuid4(), uuid.uuid4(), "a", "A", "password", ())
    with pytest.raises(ApiError) as e:
        signatures.sign(
            None,
            FakeDirectory(),
            actor,
            factor="sms",
            secret="1",
            action="x",
            meaning=None,
            settings=Settings(_env_file=None, stack="ci"),
        )
    assert e.value.code == "bad_factor"