UmbraTrans/umbratrans_web/tests/test_security.py

"""Password hashing, token handling, redaction, and policy tests."""

from __future__ import annotations

import logging

import pytest

from umbratrans_web.errors import PasswordPolicyError
from umbratrans_web.security import (
    ARGON2_MEMORY_COST_KIB,
    ARGON2_PARALLELISM,
    ARGON2_TIME_COST,
    PasswordPolicy,
    PasswordService,
    SecretRedactionFilter,
    generate_token,
    hash_session_token,
    redact_text,
    sanitize_audit_detail,
    tokens_match,
)

from .conftest import random_password


def test_production_profile_is_argon2id_with_reviewed_parameters() -> None:
    service = PasswordService()
    parameters = service.parameters()
    assert parameters["type"] == "argon2id"
    assert parameters["memory_cost_kib"] == ARGON2_MEMORY_COST_KIB
    assert parameters["time_cost"] == ARGON2_TIME_COST
    assert parameters["parallelism"] == ARGON2_PARALLELISM

    password = random_password()
    password_hash = service.hash_password(password)

    assert password_hash.startswith("$argon2id$")
    assert f"m={ARGON2_MEMORY_COST_KIB}" in password_hash
    assert f"t={ARGON2_TIME_COST}" in password_hash
    assert f"p={ARGON2_PARALLELISM}" in password_hash


def test_hash_never_contains_the_plaintext_and_verifies_once() -> None:
    service = PasswordService(
        memory_cost_kib=8 * 1024,
        time_cost=1,
        parallelism=1,
    )
    password = random_password()
    password_hash = service.hash_password(password)

    assert password not in password_hash
    assert service.verify_password(password_hash, password) is True
    assert service.verify_password(password_hash, random_password()) is False
    # A second verification of the same value still succeeds (no one-shot use).
    assert service.verify_password(password_hash, password) is True


def test_hash_is_salted_per_call() -> None:
    service = PasswordService(memory_cost_kib=8 * 1024, time_cost=1, parallelism=1)
    password = random_password()
    assert service.hash_password(password) != service.hash_password(password)


def test_verify_returns_false_for_absent_or_corrupt_hash() -> None:
    service = PasswordService(memory_cost_kib=8 * 1024, time_cost=1, parallelism=1)
    password = random_password()
    assert service.verify_password(None, password) is False
    assert service.verify_password("not-a-hash", password) is False
    assert service.needs_rehash("not-a-hash") is True


def test_repr_hides_hashes() -> None:
    service = PasswordService(memory_cost_kib=8 * 1024, time_cost=1, parallelism=1)
    password = random_password()
    password_hash = service.hash_password(password)
    assert password_hash not in repr(service)
    assert password not in repr(service)
    assert "$2" not in service.identifier(password_hash) or service.identifier(password_hash).startswith("argon2id")
    assert service.identifier(password_hash).startswith("argon2id:")


def test_password_policy_rejects_weak_candidates_without_echoing_them() -> None:
    policy = PasswordPolicy()
    with pytest.raises(PasswordPolicyError) as short_error:
        policy.validate("short", username="Umbraadmin")
    assert "min_length" in short_error.value.unmet_requirements
    assert "short" not in str(short_error.value)

    with pytest.raises(PasswordPolicyError) as reuse_error:
        policy.validate("Umbraadmin-passphrase-2026", username="Umbraadmin")
    assert "must_not_contain_username" in reuse_error.value.unmet_requirements
    assert "Umbraadmin" not in str(reuse_error.value)

    policy.validate(random_password(), username="Umbraadmin")


def test_password_policy_enforces_maximum_length() -> None:
    policy = PasswordPolicy()
    with pytest.raises(PasswordPolicyError) as error:
        policy.validate("x" * (policy.max_length + 1))
    assert "max_length" in error.value.unmet_requirements


def test_generated_temporary_passwords_satisfy_the_policy() -> None:
    policy = PasswordPolicy()
    service = PasswordService(policy=policy)
    generated = {service.generate_temporary_password(username="Umbraadmin") for _ in range(20)}

    assert len(generated) == 20
    for candidate in generated:
        policy.validate(candidate, username="Umbraadmin")
        assert len(candidate) >= policy.min_length
        assert any(character.islower() for character in candidate)
        assert any(character.isupper() for character in candidate)
        assert any(character.isdigit() for character in candidate)
        assert any(not character.isalnum() for character in candidate)
        # Ambiguous glyphs are excluded so a single transcription does not fail.
        assert not set(candidate) & set("0O1lI5S2Z")


def test_session_tokens_are_random_and_stored_only_as_a_hash() -> None:
    token = generate_token()
    other = generate_token()
    assert token != other
    assert len(token) >= 32

    token_hash = hash_session_token(token)
    assert token_hash != token
    assert token not in token_hash
    assert hash_session_token(token) == token_hash
    assert hash_session_token(other) != token_hash


def test_peppered_token_hash_differs_from_plain_hash() -> None:
    pepper = b"p" * 32
    token = generate_token()
    assert hash_session_token(token, pepper=pepper) != hash_session_token(token)
    assert hash_session_token(token, pepper=pepper) == hash_session_token(token, pepper=pepper)


def test_tokens_match_is_exact() -> None:
    assert tokens_match(generate_token(), generate_token()) is False
    value = generate_token()
    assert tokens_match(value, value) is True
    assert tokens_match(value, value + "x") is False


def test_redact_text_removes_credential_shapes() -> None:
    password = random_password()
    samples = [
        f'password="{password}"',
        "Authorization: Bearer abcdefghijklmnop",
        "$argon2id$v=19$m=65536,t=3,p=4$abcdef$0123456789abcdef",
        "-----BEGIN PRIVATE KEY-----\nabc\n-----END PRIVATE KEY-----",
    ]
    for sample in samples:
        redacted = redact_text(sample)
        assert "[redacted]" in redacted
        assert password not in redacted


def test_sanitize_audit_detail_drops_unknown_keys_and_bounds_values() -> None:
    detail = sanitize_audit_detail(
        {
            "username": "Umbraadmin",
            "status": "suspended",
            "must_change_password": True,
            "revoked_sessions": 2,
            "reason": "x" * 500,
            "password": "should-never-survive",
            "session_token": "should-never-survive",
        }
    )
    assert detail["username"] == "Umbraadmin"
    assert detail["status"] == "suspended"
    assert detail["must_change_password"] == "true"
    assert detail["revoked_sessions"] == "2"
    assert len(detail["reason"]) == 200
    assert "password" not in detail
    assert "session_token" not in detail
    assert "should-never-survive" not in "".join(detail.values())


def test_sanitize_audit_detail_redacts_secret_shaped_values() -> None:
    password = random_password()
    detail = sanitize_audit_detail({"reason": f"password={password}", "username": "Umbraadmin"})
    assert password not in detail["reason"]
    assert detail["username"] == "Umbraadmin"


def test_logging_filter_scrubs_record_arguments() -> None:
    password = random_password()
    record = logging.LogRecord(
        name="umbratrans_web.service",
        level=logging.INFO,
        pathname=__file__,
        lineno=1,
        msg="account created password=%s",
        args=(password,),
        exc_info=None,
    )
    assert SecretRedactionFilter().filter(record) is True
    assert password not in record.getMessage()
    assert "[redacted]" in record.getMessage()