# -*- coding: utf-8 -*- """``NacefClient`` — the interface over the five S-MDF web services, plus a real HTTP implementation and an in-memory mock. Topology note (deployment "Caisse en mode Cloud"): Per the CDC (§II.2 "Authentification des S-MDF agents"), the S-MDF only accepts calls from a registered *S-MDF Agent* (identified by MAC, HTTPS-only). In the cloud deployment the authorised caller is the on-till Agent, so the *live* signing path is OWL POS frontend -> local S-MDF Agent -> central S-MDF Client. ``AgentRestNacefClient`` therefore targets the Agent's base URL. It is used server-side for server-relay variants and tests; the browser-side call path is implemented in ``nacef_pos`` once the Agent's localhost/CORS contract is known. """ from __future__ import annotations import abc import base64 import datetime import json import urllib.error import urllib.request import uuid from typing import Optional from .enums import ( CertRequestStatus, SMDFConnectionState, SMDFStatus, SMDFType, ) from .errors import NacefError from .models import ( SICCertificateRequest, SICLogEntry, SMDFCertificateInfo, SMDFManifest, SMDFSignedTicket, SMDFSyncRequest, SMDFTicketInfo, SMDFVersionsInfo, EquipmentVersionsInfo, ) class NacefClient(abc.ABC): """Abstract S-MDF client. All calls raise :class:`NacefError` on failure.""" @abc.abstractmethod def get_manifest(self) -> SMDFManifest: """GET /manifest/ — current S-MDF and certificate-request state.""" @abc.abstractmethod def request_certificate(self, req: SICCertificateRequest) -> SMDFManifest: """POST /certificate/request/ — start/continue the certificate request.""" @abc.abstractmethod def sync(self, req: SMDFSyncRequest) -> SMDFManifest: """POST /sync/request/ — authenticate the taxpayer / reach SYNCHRONIZED.""" @abc.abstractmethod def sign_ticket(self, info: SMDFTicketInfo) -> SMDFSignedTicket: """POST /signature/request/ — sign a ticket, return fiscal id + QR.""" @abc.abstractmethod def log(self, entry: SICLogEntry) -> str: """POST /log/ — record a piste-d'audit entry.""" # --------------------------------------------------------------------------- # # Real HTTP client (stdlib urllib; no third-party dependency) # --------------------------------------------------------------------------- # class AgentRestNacefClient(NacefClient): """Calls the S-MDF Agent over HTTP(S) using the standard library only. Endpoint paths and the default port come from the official S-MDF OpenAPI spec (developers.nacef.tn, ``nacef-smdf-api-1.2.0``): the Agent listens on ``http://localhost:10006`` and exposes the services under ``/sic/external``. """ #: default base URL of the on-till S-MDF Agent (OpenAPI ``servers``) DEFAULT_BASE_URL = "http://localhost:10006" #: service paths (must match the official Postman collection EXACTLY, #: including trailing slashes — the S-MDF 404s/redirects otherwise). PATHS = { "manifest": "/sic/external/manifest", "certificate": "/sic/external/certificate/request/", "sync": "/sic/external/sync/request/", "sign": "/sic/external/sign/request/", "log": "/sic/external/log/", } def __init__(self, base_url: str = None, timeout: float = 20.0, verify_ssl: bool = True): self.base_url = (base_url or self.DEFAULT_BASE_URL).rstrip("/") self.timeout = timeout self.verify_ssl = verify_ssl # -- envelope handling -------------------------------------------------- # def _request(self, method: str, path: str, payload: Optional[dict]) -> dict: url = f"{self.base_url}{path}" data = None headers = {"Accept": "application/json"} if payload is not None: data = json.dumps(payload).encode("utf-8") headers["Content-Type"] = "application/json" req = urllib.request.Request(url, data=data, headers=headers, method=method) ctx = None if url.startswith("https") and not self.verify_ssl: import ssl ctx = ssl.create_default_context() ctx.check_hostname = False ctx.verify_mode = ssl.CERT_NONE try: with urllib.request.urlopen(req, timeout=self.timeout, context=ctx) as resp: body = resp.read().decode("utf-8") or "{}" return self._unwrap(resp.status, json.loads(body)) except urllib.error.HTTPError as exc: body = exc.read().decode("utf-8") or "{}" try: return self._unwrap(exc.code, json.loads(body)) except json.JSONDecodeError: raise NacefError(code=exc.code, message=body) from exc except urllib.error.URLError as exc: # network failure reaching the agent/client raise NacefError(name="SMDF_NOT_REACHABLE", message=str(exc.reason)) from exc @staticmethod def _unwrap(http_status: int, envelope: dict) -> dict: """Envelope is ``{object, errorCode, message}`` (OpenAPI ``Response``). ``errorCode`` is an integer in the live API; older/symbolic string codes are also accepted for resilience. """ error_code = envelope.get("errorCode") message = envelope.get("message") if http_status == 200 and not error_code: return envelope.get("object", envelope.get("objet", {})) or {} if isinstance(error_code, int): raise NacefError(code=error_code, message=message) if error_code: raise NacefError(name=error_code, message=message) # No envelope errorCode (e.g. a bare 401/403 from the SIC security # filter, which returns an empty body): surface the HTTP status so # callers can tell "reachable but unauthorized" from "unreachable". raise NacefError(code=http_status, message=message or "HTTP %s" % http_status) # -- interface ---------------------------------------------------------- # def get_manifest(self) -> SMDFManifest: return SMDFManifest.from_dict( self._request("GET", self.PATHS["manifest"], None)) def request_certificate(self, req: SICCertificateRequest) -> SMDFManifest: return SMDFManifest.from_dict( self._request("POST", self.PATHS["certificate"], req.to_dict())) def sync(self, req: SMDFSyncRequest) -> SMDFManifest: return SMDFManifest.from_dict( self._request("POST", self.PATHS["sync"], req.to_dict())) def sign_ticket(self, info: SMDFTicketInfo) -> SMDFSignedTicket: return SMDFSignedTicket.from_dict( self._request("POST", self.PATHS["sign"], info.to_dict())) def log(self, entry: SICLogEntry) -> str: self._request("POST", self.PATHS["log"], entry.to_dict()) return "OK" # --------------------------------------------------------------------------- # # In-memory mock — drives the NACEF-PROCTEST-02 21-step scenario offline # --------------------------------------------------------------------------- # class MockNacefClient(NacefClient): """Deterministic in-memory S-MDF for dev/CI and for building the POS flow. It reproduces the state machine of CDC §I: FACTORY -> CERT_REQUESTED -> (certificate generated externally) -> SYNCHRONIZED, with offline-ticket accounting and suspend/revoke/maintenance transitions. The ``simulate_*`` helpers stand in for actions that happen outside the caisse (registration unit, platform admin). """ def __init__(self, imdf: str = "IMDF00000000001", smdf_type: SMDFType = SMDFType.SERVER, version: str = "1.0.0", offline_capacity: int = 5): self.version = version self.offline_capacity = offline_capacity self._counter = 0 self._signed = [] # list of SMDFSignedTicket self.logs = [] # list of SICLogEntry self.manifest = SMDFManifest( imdf=imdf, status=SMDFStatus.FACTORY.value, version=version, type=smdf_type.value, certificate_info=SMDFCertificateInfo(cert_request_status=""), maintenance_enabled=False, state=SMDFConnectionState.ONLINE.value, synchronization_rate=0, version_info=SMDFVersionsInfo( smdf=EquipmentVersionsInfo(os="Linux", current=version, last=version, minimal=version), agents={}), available_offline_tickets=0, ) # -- connectivity toggles (test control) -------------------------------- # def set_online(self): self.manifest.state = SMDFConnectionState.ONLINE.value def set_offline(self): self.manifest.state = SMDFConnectionState.OFFLINE.value @property def is_online(self) -> bool: return self.manifest.state == SMDFConnectionState.ONLINE.value # -- external-world simulation ------------------------------------------ # def simulate_certificate_generation(self, valid_days: int = 365): """Registration unit issues the certificate (PROCTEST step 7).""" now = datetime.datetime.now(datetime.timezone.utc).replace(tzinfo=None) ci = self.manifest.certificate_info ci.cert_request_status = CertRequestStatus.CERTIFICATE_GENERATED.value ci.issuance_date = now.isoformat(timespec="milliseconds") ci.expiration_date = (now + datetime.timedelta(days=valid_days) ).isoformat(timespec="milliseconds") ci.expired = False ci.revoked = False def simulate_suspend(self): self.manifest.status = SMDFStatus.SUSPENDED.value def simulate_maintenance(self): self.manifest.maintenance_enabled = True self.manifest.status = SMDFStatus.MAINTENANCE.value def simulate_revoke(self): self.manifest.certificate_info.revoked = True def simulate_expire(self): self.manifest.certificate_info.expired = True # -- NacefClient interface ---------------------------------------------- # def get_manifest(self) -> SMDFManifest: return self.manifest def request_certificate(self, req: SICCertificateRequest) -> SMDFManifest: if self.manifest.certificate_info.revoked or \ self.manifest.certificate_info.expired: # allow a fresh request after revoke/expire (PROCTEST 21) self.manifest.certificate_info = SMDFCertificateInfo(cert_request_status="") if self.manifest.status not in (SMDFStatus.FACTORY.value,): if self.manifest.certificate_info.cert_request_status == \ CertRequestStatus.CERTIFICATE_GENERATED.value: raise NacefError(name="SMDF_ALREADY_HAS_CERTIFICATE") self.manifest.status = SMDFStatus.CERT_REQUESTED.value self.manifest.certificate_info.cert_request_status = \ CertRequestStatus.PIN_VALIDATED.value return self.manifest def sync(self, req: SMDFSyncRequest) -> SMDFManifest: ci = self.manifest.certificate_info if ci.revoked: raise NacefError(name="SMDF_REVOKED_CERTIFICATE") if ci.expired: raise NacefError(name="SMDF_EXPIRED_CERTIFICATE") if ci.cert_request_status != CertRequestStatus.CERTIFICATE_GENERATED.value: raise NacefError(name="SMDF_CERTIFICATE_NOT_GENERATED") # successful synchronisation self.manifest.status = SMDFStatus.SYNCHRONIZED.value self.manifest.maintenance_enabled = False self.manifest.synchronization_rate = 100 self.manifest.available_offline_tickets = self.offline_capacity # first sync returns the "Ticket 0" report (CDC §I.4.d) self.manifest.ticket_zero_qrcode = base64.b64encode( self._ticket_zero_text().encode("utf-8")).decode("ascii") return self.manifest def sign_ticket(self, info: SMDFTicketInfo) -> SMDFSignedTicket: ci = self.manifest.certificate_info if ci.revoked: raise NacefError(name="SMDF_REVOKED_CERTIFICATE") if ci.expired: raise NacefError(name="SMDF_EXPIRED_CERTIFICATE") if self.manifest.status != SMDFStatus.SYNCHRONIZED.value: raise NacefError(name="SMDF_NOT_SYNCHRONIZED") if not self.is_online: if self.manifest.available_offline_tickets <= 0: # offline budget exhausted -> force resync (PROCTEST 14/15) self.manifest.status = SMDFStatus.NOT_SYNCHRONIZED.value raise NacefError(name="SMDF_NOT_SYNCHRONIZED") self.manifest.available_offline_tickets -= 1 self._counter += 1 mode = "ONLINE" if self.is_online else "OFFLINE" # unique per signature (the real S-MDF assigns a globally unique id; the # mock is stateless per call, so add a short random suffix) ticket_id = f"{self.manifest.imdf}-{uuid.uuid4().hex[:12].upper()}" qr_lines = [mode] if mode == "OFFLINE": qr_lines += [self.manifest.imdf, ticket_id, info.operation_type, info.transaction_type, str(info.total_ht), str(info.total_tax), uuid.uuid4().hex] else: qr_lines += [ticket_id] qr = base64.b64encode("\n".join(qr_lines).encode("utf-8")).decode("ascii") signed = SMDFSignedTicket(ticket_identifier=ticket_id, qrcode_image=qr) self._signed.append(signed) return signed def log(self, entry: SICLogEntry) -> str: self.logs.append(entry) return "OK" # -- helpers ------------------------------------------------------------ # def _ticket_zero_text(self) -> str: return ( "Bienvenue dans la plateforme NACEF. Votre SMDF a ete configure " "correctement.\n" f"NACEF SMDF {self.version}\n{self.manifest.imdf}\n" f"{self.manifest.status}\n{self.manifest.type}\n{self.manifest.state}\n" f"Counter: {self._counter}\n" f"{datetime.datetime.now(datetime.timezone.utc).replace(tzinfo=None).isoformat(timespec='milliseconds')}" )