GTI_GERENCIADOR_FISCAL/fiscal/agent/tests/test_credential_store.py

"""DPAPI LocalMachine credential store: encrypt/decrypt round-trip, plaintext
fallback is dev-only, ACL is SYSTEM/Administrators, and the token is never
written into agent.env."""
import importlib.util
import os
import re
import stat
import subprocess
import sys
from pathlib import Path

import pytest

from gti_fiscal_agent import credential_store

import win32api

AGENT_ROOT = Path(__file__).resolve().parents[1]
RUN_SCRIPT = AGENT_ROOT / "scripts" / "run_agent_service.py"


def load_run_service():
    spec = importlib.util.spec_from_file_location("run_agent_service_cred", RUN_SCRIPT)
    module = importlib.util.module_from_spec(spec)
    sys.modules[spec.name] = module
    spec.loader.exec_module(module)
    return module


def write_plaintext(path: Path, value: str = "TOKEN_PLAINTEXT_DEV"):
    path.write_text(value, encoding="utf-8")


# ---------- DPAPI round-trip ----------

def _allow_current_user(path: Path) -> None:
    """The strict SYSTEM/Administrators DACL blocks the pytest user (not an
    admin here). As the file OWNER the current user may grant itself read —
    exactly like the elevated installer/local administrator does on the real
    host before the credential is consumed by the SYSTEM service."""
    import subprocess

    import win32security

    acct_sid, _domain, _type = win32security.LookupAccountName(None, win32api.GetUserName())
    sid_str = win32security.ConvertSidToStringSid(acct_sid)
    subprocess.run(["icacls", str(path), "/grant", f"*{sid_str}:(R)"], check=True, capture_output=True)


def test_write_then_read_roundtrip(tmp_path):
    path = tmp_path / "credentials.bin"
    credential_store.write_credential(path, "TOKEN_DPAPI_1")
    _allow_current_user(path)
    assert credential_store.read_credential(path) == "TOKEN_DPAPI_1"


def test_stored_blob_never_contains_the_plaintext_secret(tmp_path):
    path = tmp_path / "credentials.bin"
    credential_store.write_credential(path, "TOKEN_VERY_SECRET_9")
    _allow_current_user(path)
    raw = path.read_bytes()
    assert b"TOKEN_VERY_SECRET_9" not in raw


def test_read_missing_file_raises(tmp_path):
    with pytest.raises(credential_store.CredentialStoreError):
        credential_store.read_credential(tmp_path / "nope.bin")


def test_write_detects_dpapi_unavailable_and_fails_closed(tmp_path, monkeypatch):
    monkeypatch.setattr(credential_store, "_dpapi_available", lambda: False)
    with pytest.raises(credential_store.CredentialStoreError):
        credential_store.write_credential(tmp_path / "credentials.bin", "never")


# ---------- plaintext fallback (dev-only) ----------

def test_plaintext_rejected_by_default(tmp_path):
    path = tmp_path / "credentials.bin"
    write_plaintext(path)
    with pytest.raises(credential_store.CredentialStoreError):
        credential_store.read_credential(path)


def test_plaintext_accepted_only_with_dev_flag(tmp_path, monkeypatch):
    path = tmp_path / "credentials.bin"
    write_plaintext(path, "TOKEN_DEV_PLAIN")
    assert credential_store.read_credential(path, env={"GTI_ALLOW_PLAINTEXT_CREDENTIAL": "1"}) == "TOKEN_DEV_PLAIN"
    monkeypatch.setenv("GTI_ALLOW_PLAINTEXT_CREDENTIAL", "1")
    assert credential_store.read_credential(path) == "TOKEN_DEV_PLAIN"
    monkeypatch.delenv("GTI_ALLOW_PLAINTEXT_CREDENTIAL", raising=False)
    with pytest.raises(credential_store.CredentialStoreError):
        credential_store.read_credential(path)


# ---------- ACL (SYSTEM/Administrators only) ----------

def test_write_sets_system_and_administrators_acl(tmp_path):
    import win32security

    path = tmp_path / "credentials.bin"
    credential_store.write_credential(path, "TOKEN_ACL_7")
    admin_sid = win32security.ConvertStringSidToSid("S-1-5-32-544")
    system_sid = win32security.ConvertStringSidToSid("S-1-5-18")
    sd = win32security.GetFileSecurity(str(path), win32security.DACL_SECURITY_INFORMATION)
    dacl = sd.GetSecurityDescriptorDacl()
    assert dacl.GetAceCount() == 2
    allow_trustees = []
    for i in range(dacl.GetAceCount()):
        ace = dacl.GetAce(i)
        assert ace[0][0] == win32security.ACCESS_ALLOWED_ACE_TYPE, ace  # only allows, no deny
        allow_trustees.append(win32security.ConvertSidToStringSid(ace[2]))
    assert set(allow_trustees) == {"S-1-5-18", "S-1-5-32-544"}


# ---------- install_env_from_credentials / Config integration ----------

def test_install_env_from_credentials_fills_agent_token(tmp_path):
    path = tmp_path / "credentials.bin"
    credential_store.write_credential(path, "TOKEN_FROM_FILE")
    _allow_current_user(path)
    env = {"AGENT_TOKEN": "", "HUB_URL": "https://x"}
    install_env_from_credentials = credential_store.install_env_from_credentials
    assert install_env_from_credentials(path, env) == "TOKEN_FROM_FILE"
    assert env["AGENT_TOKEN"] == "TOKEN_FROM_FILE"


def test_install_env_prefers_explicit_agent_token(tmp_path):
    path = tmp_path / "credentials.bin"
    credential_store.write_credential(path, "TOKEN_FROM_FILE")
    env = {"AGENT_TOKEN": "TOKEN_FROM_ENV"}
    install_env_from_credentials = credential_store.install_env_from_credentials
    assert install_env_from_credentials(path, env) == "TOKEN_FROM_ENV"
    assert env["AGENT_TOKEN"] == "TOKEN_FROM_ENV"


def test_config_reads_token_from_credential_file(tmp_path, monkeypatch):
    credential_path = tmp_path / "credentials.bin"
    credential_store.write_credential(credential_path, "TOKEN_CRED_FILE")
    _allow_current_user(credential_path)
    env = {
        "FIREBIRD_DATABASE_PATH": str(tmp_path / "nope.fdb"),
        "FIREBIRD_WRITES": "0",
        "FISCAL_MODE": "DRY_RUN",
        "GTI_CREDENTIAL_FILE": str(credential_path),
    }
    for key in list(env):
        monkeypatch.setenv(key, env[key])
    service = load_run_service()
    config = service.Config(os.environ)
    assert config.agent_token == "TOKEN_CRED_FILE"
    assert config.firebird_writes is False  # FIREBIRD_WRITES stays 0 regardless of credential


# ---------- no-write enforcement (drift) ----------

def test_firebird_writes_fail_closed_with_credential_file(tmp_path, monkeypatch):
    credential_path = tmp_path / "credentials.bin"
    credential_store.write_credential(credential_path, "TOKEN")
    _allow_current_user(credential_path)
    env = {
        "FIREBIRD_DATABASE_PATH": str(tmp_path / "x.fdb"),
        "FIREBIRD_WRITES": "0",
        "GTI_CREDENTIAL_FILE": str(credential_path),
    }
    for key in list(env):
        monkeypatch.setenv(key, env[key])
    service = load_run_service()

    # Even a hostile FISCAL_MODE cannot enable writes when FIREBIRD_WRITES=0.
    monkeypatch.setenv("FISCAL_MODE", "AUTO_APPLY")
    config = service.Config(os.environ)
    assert config.firebird_writes is False


def test_credential_file_default_lives_under_program_files():
    default = credential_store.default_credential_path()
    assert "Program Files" in str(default)
    assert default.name == "credentials.bin"