dodil/erp-brick-tax
DODIL ERP brick: tax
src/brick_tax/infrastructure/hmrc.py
"""HMRC's VAT (MTD) API behind ONE port, and the platform secret store behind another. The domain never knows how HMRC is
reached; a test swaps in an in-memory fake.
Base URLs by environment (the owner's control `hmrc_environment`): sandbox https://test-api.service.hmrc.gov.uk, production
https://api.service.hmrc.gov.uk. OAuth: the owner signs in to the Government Gateway (authorization code, scopes read:vat
write:vat); the application id and secret are this deployment's (HMRC_CLIENT_ID / HMRC_CLIENT_SECRET). The tokens are kept ONLY
in the secret store (a mounted secret directory, SECRETS_DIR, one file per reference) — never in a row, never in a log.
"""
import json
import os
import pathlib
from datetime import date
from typing import Optional, Protocol
from urllib.parse import urlencode
import httpx
BASES = {"sandbox": "https://test-api.service.hmrc.gov.uk", "production": "https://api.service.hmrc.gov.uk"}
TIMEOUT = 20.0
SCOPES = "read:vat write:vat"
class HmrcError(Exception):
"""HMRC answered with an error: its code (DUPLICATE_SUBMISSION, VRN_INVALID ...) and message."""
def __init__(self, code: str, message: str = "", status: int = 0):
super().__init__(f"{code}: {message}" if message else code)
self.code, self.message, self.status = code, message, status
class HmrcTimeout(HmrcError):
"""No answer: HMRC may or may not have the request. Settle with a read before any retry."""
def __init__(self):
super().__init__("TIMEOUT", "HMRC did not answer in time")
class HmrcPort(Protocol):
def authorize_url(self, environment: str, state: str, redirect_uri: str) -> str: ...
def exchange_code(self, environment: str, code: str, redirect_uri: str) -> dict:
"""{access_token, refresh_token, expires_in}."""
def obligations(self, environment: str, vrn: str, frm: date, to: date, tokens: dict, headers: dict) -> list:
"""GET /organisations/vat/{vrn}/obligations -> [{periodKey, start, end, due, status O|F, received?}]."""
def get_return(self, environment: str, vrn: str, period_key: str, tokens: dict, headers: dict) -> Optional[dict]:
"""GET /organisations/vat/{vrn}/returns/{periodKey} -> HMRC's figures, or None when HMRC has no such return."""
def submit_return(self, environment: str, vrn: str, body: dict, tokens: dict, headers: dict) -> dict:
"""POST /organisations/vat/{vrn}/returns -> {processingDate, formBundleNumber, paymentIndicator, chargeRefNumber}."""
def liabilities(self, environment: str, vrn: str, frm: date, to: date, tokens: dict, headers: dict) -> list:
"""GET /organisations/vat/{vrn}/liabilities -> [{taxPeriod:{from,to}, type, originalAmount, outstandingAmount, due}]."""
def payments(self, environment: str, vrn: str, frm: date, to: date, tokens: dict, headers: dict) -> list:
"""GET /organisations/vat/{vrn}/payments -> [{amount, received}]."""
def check_vat_number(self, environment: str, vat_number: str) -> Optional[dict]:
"""HMRC's 'check a UK VAT number' lookup -> {valid, name, address, consultation_number}, or None when HMRC does not know it."""
def verify_subcontractor(self, environment: str, body: dict, tokens: dict, headers: dict) -> dict:
"""CIS verification -> {status gross | net | unmatched, verification_number}."""
def submit_cis300(self, environment: str, body: dict, tokens: dict, headers: dict) -> dict:
"""CIS300 monthly return -> HMRC's receipt {correlation_id, accepted_at}."""
class SecretsPort(Protocol):
def put(self, ref: str, value: dict) -> None: ...
def get(self, ref: str) -> Optional[dict]: ...
class HttpHmrc:
def _creds(self) -> tuple:
return os.environ.get("HMRC_CLIENT_ID", ""), os.environ.get("HMRC_CLIENT_SECRET", "")
def authorize_url(self, environment: str, state: str, redirect_uri: str) -> str:
q = urlencode({"response_type": "code", "client_id": self._creds()[0], "scope": SCOPES, "state": state, "redirect_uri": redirect_uri})
return f"{BASES[environment]}/oauth/authorize?{q}"
def exchange_code(self, environment: str, code: str, redirect_uri: str) -> dict:
cid, secret = self._creds()
try:
r = httpx.post(f"{BASES[environment]}/oauth/token", timeout=TIMEOUT, data={
"grant_type": "authorization_code", "client_id": cid, "client_secret": secret, "code": code, "redirect_uri": redirect_uri})
except httpx.TimeoutException:
raise HmrcTimeout() from None
except httpx.HTTPError as exc:
raise HmrcError("UNREACHABLE", str(exc)[:200]) from None
if r.status_code >= 300:
raise HmrcError("OAUTH_REFUSED", r.text[:200], r.status_code)
return r.json()
def _call(self, method: str, environment: str, path: str, tokens: dict, headers: dict, *, params=None, body=None):
hdrs = {"Accept": "application/vnd.hmrc.1.0+json", "Authorization": f"Bearer {tokens.get('access_token', '')}", **headers}
if body is not None:
hdrs["Content-Type"] = "application/json"
try:
r = httpx.request(method, f"{BASES[environment]}{path}", params=params, headers=hdrs, timeout=TIMEOUT,
content=None if body is None else json.dumps(body, default=float))
except httpx.TimeoutException:
raise HmrcTimeout() from None
except httpx.HTTPError as exc:
raise HmrcError("UNREACHABLE", str(exc)[:200]) from None
if r.status_code == 404 and method == "GET":
return None
if r.status_code >= 300:
try:
err = r.json()
except ValueError:
err = {}
raise HmrcError(err.get("code") or f"HTTP_{r.status_code}", err.get("message") or r.text[:200], r.status_code)
return r.json() if r.content else {}
def obligations(self, environment, vrn, frm, to, tokens, headers):
out = self._call("GET", environment, f"/organisations/vat/{vrn}/obligations", tokens, headers,
params={"from": frm.isoformat(), "to": to.isoformat()})
return (out or {}).get("obligations", [])
def get_return(self, environment, vrn, period_key, tokens, headers):
return self._call("GET", environment, f"/organisations/vat/{vrn}/returns/{period_key}", tokens, headers)
def submit_return(self, environment, vrn, body, tokens, headers):
return self._call("POST", environment, f"/organisations/vat/{vrn}/returns", tokens, headers, body=body)
def liabilities(self, environment, vrn, frm, to, tokens, headers):
out = self._call("GET", environment, f"/organisations/vat/{vrn}/liabilities", tokens, headers,
params={"from": frm.isoformat(), "to": to.isoformat()})
return (out or {}).get("liabilities", [])
def payments(self, environment, vrn, frm, to, tokens, headers):
out = self._call("GET", environment, f"/organisations/vat/{vrn}/payments", tokens, headers,
params={"from": frm.isoformat(), "to": to.isoformat()})
return (out or {}).get("payments", [])
def check_vat_number(self, environment, vat_number):
"""GET /organisations/vat/check-vat-number/lookup/{target}: an application-restricted API (HMRC_SERVER_TOKEN)."""
out = self._call("GET", environment, f"/organisations/vat/check-vat-number/lookup/{vat_number}",
{"access_token": os.environ.get("HMRC_SERVER_TOKEN", "")}, {})
if out is None:
return None
t = out.get("target") or {}
addr = t.get("address") or {}
return {"valid": True, "name": t.get("name") or "", "address": ", ".join(str(v) for v in addr.values() if v),
"consultation_number": out.get("consultationNumber") or ""}
def verify_subcontractor(self, environment, body, tokens, headers):
# HMRC offers CIS verification and the CIS300 only through the Government Gateway XML channel (GovTalk), not the REST
# API this adapter speaks. Until that channel is wired a person reads the result from HMRC's CIS service and enters it.
raise HmrcError("CIS_CHANNEL_NOT_WIRED", "CIS verification runs through HMRC's Government Gateway XML channel, which is "
"not connected: read the result from HMRC's CIS service and enter it with its verification number")
def submit_cis300(self, environment, body, tokens, headers):
raise HmrcError("CIS_CHANNEL_NOT_WIRED", "the CIS300 is filed through HMRC's Government Gateway XML channel, which is not "
"connected: file it in HMRC's CIS online service and record the receipt")
class FileSecrets:
"""The platform's secret store as a mounted directory: one 0600 file per reference. Unset SECRETS_DIR = no store = no
connection can be made (the owner is told), never a token kept somewhere else."""
def _path(self, ref: str) -> pathlib.Path:
base = os.environ.get("SECRETS_DIR", "")
if not base:
raise HmrcError("NO_SECRET_STORE", "SECRETS_DIR is not set: the HMRC tokens have nowhere safe to live")
return pathlib.Path(base) / (ref.replace("/", "__") + ".json")
def put(self, ref: str, value: dict) -> None:
p = self._path(ref)
p.parent.mkdir(parents=True, exist_ok=True)
p.write_text(json.dumps(value))
p.chmod(0o600)
def get(self, ref: str) -> Optional[dict]:
p = self._path(ref)
return json.loads(p.read_text()) if p.exists() else None