installer v1 (A+B): install-code enrollment, panel code UI, EXE publish; DPAPI credential store
This commit is contained in:
gti
@@ -76,7 +76,11 @@ class Config:
|
||||
def __init__(self, env):
|
||||
self.hub_url = env.get("GTI_HUB_URL") or env.get("HUB_URL") or "http://localhost:8000"
|
||||
self.agent_id = env.get("AGENT_ID", "")
|
||||
self.agent_token = env.get("AGENT_TOKEN") or env.get("AGENT_DEV_TOKEN", "")
|
||||
self.credential_file = env.get("GTI_CREDENTIAL_FILE") or str(Path(os.environ.get("PROGRAMFILES", r"C:\Program Files")) / "GTI" / "FiscalAgent" / "credentials.bin")
|
||||
# The permanent token NEVER lives in agent.env: it is decrypted from
|
||||
# credentials.bin (DPAPI LocalMachine + SYSTEM/Administrators ACL). An
|
||||
# explicit AGENT_TOKEN/AGENT_DEV_TOKEN in env wins (dev override only).
|
||||
self.agent_token = self._resolve_agent_token(env)
|
||||
self.company_id = env.get("COMPANY_ID", "")
|
||||
self.company_uf = env.get("COMPANY_UF", "")
|
||||
self.database_path = env["FIREBIRD_DATABASE_PATH"]
|
||||
@@ -98,6 +102,22 @@ class Config:
|
||||
self.expected_schema_fingerprint = env.get("EXPECTED_SCHEMA_FINGERPRINT", REFERENCE_SCHEMA_V1)
|
||||
self.log_path = env.get("AGENT_LOG_PATH", str(AGENT_ROOT / "logs" / "agent.log"))
|
||||
|
||||
@staticmethod
|
||||
def _resolve_agent_token(env):
|
||||
"""DPAPI credential from credentials.bin; explicit AGENT_TOKEN env wins
|
||||
(dev override). Fail closed: never log or fabricate a token."""
|
||||
from gti_fiscal_agent.credential_store import install_env_from_credentials
|
||||
|
||||
explicit = env.get("AGENT_TOKEN") or env.get("AGENT_DEV_TOKEN") or ""
|
||||
if explicit:
|
||||
return explicit
|
||||
try:
|
||||
return install_env_from_credentials(env.get("GTI_CREDENTIAL_FILE") or str(
|
||||
Path(os.environ.get("PROGRAMFILES", r"C:\Program Files")) / "GTI" / "FiscalAgent" / "credentials.bin"), env)
|
||||
except Exception as exc: # noqa: BLE001 — fail closed: no token, no heartbeat
|
||||
logger.warning("no usable agent credential: %s", exc)
|
||||
return ""
|
||||
|
||||
|
||||
class GracefulStop:
|
||||
def __init__(self):
|
||||
|
||||
@@ -0,0 +1,137 @@
|
||||
"""Windows DPAPI credential store for the GTI Fiscal Agent.
|
||||
|
||||
The permanent Agent credential (AGENT_TOKEN) must NEVER live in `agent.env`
|
||||
(which is world-readable config). It is stored in a dedicated
|
||||
`credentials.bin` file, encrypted with DPAPI in LocalMachine scope
|
||||
(`CRYPTPROTECT_LOCAL_MACHINE`), so any local Windows account on the host can
|
||||
decrypt it — and the file ACL is tightened to SYSTEM/Administrators only via
|
||||
`icacls`. The token is decrypted in-process at runtime by the service host and
|
||||
never logged.
|
||||
|
||||
`write_credential`/`read_credential` are shared by Setup.exe (writes at
|
||||
install time) and the Service host (reads at every start).
|
||||
|
||||
Fallback: a plaintext token file is accepted ONLY in development, guarded by
|
||||
`GTI_ALLOW_PLAINTEXT_CREDENTIAL=1`. Production Setup.exe never enables it.
|
||||
"""
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
_PROGRAM_DIR = Path("C:/Program Files/GTI/FiscalAgent")
|
||||
_MAGIC = b"GTIDPAPI1"
|
||||
_DEFAULT_FLAGS = 0x1 | 0x4 # CRYPTPROTECT_UI_FORBIDDEN | CRYPTPROTECT_LOCAL_MACHINE
|
||||
|
||||
|
||||
class CredentialStoreError(Exception):
|
||||
pass
|
||||
|
||||
|
||||
class _DpapiLazy:
|
||||
"""Lazy wrapper around win32crypt so importing this module (and the whole
|
||||
agent package) works on non-Windows hosts without pywin32. Functions accept
|
||||
the same signatures pywin32 exposes on this machine's build, tolerating both
|
||||
the tuple and the bytes return shapes seen across pywin32 versions."""
|
||||
|
||||
def __init__(self):
|
||||
self._crypt = None
|
||||
|
||||
def _api(self):
|
||||
if self._crypt is None:
|
||||
try:
|
||||
import win32crypt
|
||||
except Exception as exc: # noqa: BLE001
|
||||
raise CredentialStoreError(f"win32crypt unavailable: {exc}") from exc
|
||||
self._crypt = win32crypt
|
||||
return self._crypt
|
||||
|
||||
def protect(self, data):
|
||||
crypt = self._api()
|
||||
returned = crypt.CryptProtectData(data, "gti-fiscal-agent-credential", None, None, None, _DEFAULT_FLAGS)
|
||||
if isinstance(returned, tuple):
|
||||
returned, _ = returned
|
||||
return bytes(returned)
|
||||
|
||||
def unprotect(self, blob):
|
||||
crypt = self._api()
|
||||
returned = crypt.CryptUnprotectData(bytes(blob), None, None, None, 0)
|
||||
if isinstance(returned, tuple):
|
||||
_, returned = returned
|
||||
return bytes(returned)
|
||||
|
||||
|
||||
_dpapi = _DpapiLazy()
|
||||
|
||||
|
||||
def _dpapi_available() -> bool:
|
||||
try:
|
||||
_dpapi._api()
|
||||
return True
|
||||
except CredentialStoreError:
|
||||
return False
|
||||
|
||||
|
||||
def default_credential_path() -> Path:
|
||||
return _PROGRAM_DIR / "credentials.bin"
|
||||
|
||||
|
||||
def _apply_acl(path: Path) -> None:
|
||||
"""Tighten ACL so only SYSTEM and Administrators can read/modify the blob.
|
||||
Well-known SIDs are used (`*S-1-5-18` = SYSTEM, `*S-1-5-32-544` = BUILTIN\
|
||||
Administrators) because account-name mapping is locale-dependent and fails
|
||||
on non-English Windows. Setup.exe runs elevated (`--uac-admin`), so icacls
|
||||
succeeds (the owner — the installing administrator — can change the DACL)."""
|
||||
cmd = ["icacls", str(path), "/inheritance:r", "/grant:r", "*S-1-5-18:(F)", "*S-1-5-32-544:(F)"]
|
||||
try:
|
||||
result = subprocess.run(cmd, capture_output=True, text=True)
|
||||
except FileNotFoundError as exc: # non-Windows host without icacls
|
||||
raise CredentialStoreError(f"icacls unavailable: {exc}") from exc
|
||||
if result.returncode != 0:
|
||||
raise CredentialStoreError(f"icacls failed: {result.stdout} {result.stderr}")
|
||||
|
||||
|
||||
def write_credential(path, payload: str) -> None:
|
||||
"""Encrypt `payload` with DPAPI LocalMachine and write `path` with a
|
||||
SYSTEM/Administrators-only ACL. Fails closed: no plaintext file is ever
|
||||
created on this path."""
|
||||
if isinstance(payload, str):
|
||||
payload = payload.encode("utf-8")
|
||||
if not _dpapi_available():
|
||||
raise CredentialStoreError("DPAPI unavailable; refusing to store the credential as plaintext (fail closed)")
|
||||
path = Path(path)
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
blob = _MAGIC + _dpapi.protect(payload)
|
||||
path.write_bytes(blob)
|
||||
try:
|
||||
_apply_acl(path)
|
||||
except CredentialStoreError:
|
||||
path.unlink(missing_ok=True) # do not leave a credential file without its ACL
|
||||
raise
|
||||
|
||||
|
||||
def read_credential(path, *, env=None) -> str:
|
||||
"""Decrypt `path`. A plaintext file is honored ONLY when
|
||||
`GTI_ALLOW_PLAINTEXT_CREDENTIAL=1` is set in `env`/environment (development)."""
|
||||
env = env if env is not None else os.environ
|
||||
path = Path(path)
|
||||
if not path.exists():
|
||||
raise CredentialStoreError(f"credential file missing: {path}")
|
||||
raw = path.read_bytes()
|
||||
if raw.startswith(_MAGIC):
|
||||
return _dpapi.unprotect(raw[len(_MAGIC):]).decode("utf-8")
|
||||
if (env.get("GTI_ALLOW_PLAINTEXT_CREDENTIAL") or "").strip() != "1":
|
||||
raise CredentialStoreError("credential file is plaintext and GTI_ALLOW_PLAINTEXT_CREDENTIAL is not set; refusing (fail closed)")
|
||||
return raw.decode("utf-8").strip()
|
||||
|
||||
|
||||
def install_env_from_credentials(path, env) -> str:
|
||||
"""Fill `env["AGENT_TOKEN"]` from `credentials.bin` unless the env already
|
||||
carries an explicit token (which wins). Returns the effective token —
|
||||
WITHOUT writing it into `agent.env` — and never logs it."""
|
||||
existing = env.get("AGENT_TOKEN") or env.get("AGENT_DEV_TOKEN") or ""
|
||||
if existing:
|
||||
return existing
|
||||
token = read_credential(path, env=env)
|
||||
env["AGENT_TOKEN"] = token
|
||||
return token
|
||||
@@ -0,0 +1,179 @@
|
||||
"""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"
|
||||
Reference in New Issue
Block a user