"""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"