"""Les 17 contrôles de l'audit 5D, groupés en 5 dimensions. Chaque contrôle est une fonction pure `fn(art, spec) -> (statut, detail)` où `art` est le dict des livrables (voir `artifacts.py`) et `spec` la spec d'audit (blocs réglementaires déclaratifs). Trois statuts : PASS — l'invariant structurel tient. FAIL — violation : incohérence entre livrables OU valeur fabriquée (un paramètre chiffré présent SANS `source`). Un FAIL est un vrai défaut à corriger avant import VPS. A_CONFIRMER — la structure est correcte MAIS un paramètre réglementaire réel (taux, RNC, seuil UAF…) est légitimement en attente. Ce n'est pas un échec : c'est un « open item » remonté au bon métier. Anti-invention #6 : on n'invente jamais pour « faire PASS ». La méta-donnée de chaque contrôle (titre, référence normative, artefacts, propriétaire de l'open item) vit dans le registre `CONTROLS` ci-dessous ; le fichier `audit_spec.json` en est le miroir humain, et un invariant du générateur vérifie la parité des `id` entre code et spec. """ from __future__ import annotations from typing import Any, Callable from .deps import RoleResolver, is_filled, load_contract, roles_targeting PASS = "PASS" FAIL = "FAIL" A_CONFIRMER = "A_CONFIRMER" # Les 5 dimensions (5D) et la norme de référence de chacune. DIMENSIONS: list[dict[str, str]] = [ {"id": "D1", "titre": "Traçabilité & anti-invention", "reference": "ISA 500 — Éléments probants"}, {"id": "D2", "titre": "Conformité AML / UAF", "reference": "Ley 155-17 — sujeto obligado immobilier · KYC"}, {"id": "D3", "titre": "Conformité fiscale e-CF", "reference": "Ley 32-23 · DGII · Cardnet (CLAUDE.md #10)"}, {"id": "D4", "titre": "Intégrité référentielle du reporting", "reference": "IFRS — cohérence documentaire"}, {"id": "D5", "titre": "Gouvernance & ségrégation des tâches", "reference": "ISA 315 — contrôle interne (SoD)"}, ] # Clés DocPerm booléennes → vocabulaire d'action (ordre canonique, déterminisme). _PERM_KEYS = ["read", "write", "create", "submit", "cancel", "amend", "delete", "report", "print", "email", "share", "export"] # --------------------------------------------------------------------------- # # Helpers de lecture (aucune fabrication : simple projection des artefacts). # --------------------------------------------------------------------------- # def _fields_by_name(doctype: dict[str, Any]) -> dict[str, dict]: return {f["fieldname"]: f for f in doctype.get("fields", [])} def _select_options(field: dict[str, Any]) -> list[str]: return [o for o in (field.get("options") or "").split("\n") if o.strip()] def _currency_fields(doctype: dict[str, Any]) -> set[str]: return {f["fieldname"] for f in doctype.get("fields", []) if f.get("fieldtype") == "Currency"} # --------------------------------------------------------------------------- # # D1 · Traçabilité & anti-invention (ISA 500) # --------------------------------------------------------------------------- # def d1_1_commissions_taux(art, spec): evs = art["commissions"]["evenements"] bad, open_, filled = [], [], [] for e in evs: taux, src, ac = e.get("taux_pct"), e.get("source"), e.get("a_confirmer") lib = e.get("libelle", e.get("update_value")) if taux is None: (open_ if (ac is True and src is None) else bad).append(lib) else: (filled if is_filled(src) else bad).append(lib) if bad: return FAIL, f"{len(bad)} taux fabriqué(s) ou incohérent(s) : {bad}" if open_: return A_CONFIRMER, (f"{len(open_)}/{len(evs)} taux de commission à " f"confirmer (aucun fabriqué · source null).") return PASS, f"{len(filled)} taux confirmés, chacun sourcé." def d1_2_ecf_emisor(art, spec): em = art["ecf"]["emisor"] rnc, rs, src, ac = (em.get("rnc_emisor"), em.get("razon_social"), em.get("source"), em.get("a_confirmer")) if rnc is None and rs is None: if ac is True and src is None: return A_CONFIRMER, "RNC + raison sociale émetteur à confirmer (aucun fabriqué)." return FAIL, "émetteur vide mais mal marqué (a_confirmer/source incohérents)." if not is_filled(src): return FAIL, "RNC/raison sociale renseignés SANS source (fabrication)." return PASS, "émetteur renseigné avec source." def d1_3_ecf_itbis_cambio(art, spec): ecf = art["ecf"] bad, open_ = [], [] for tx in ecf["taxes"]: if tx.get("taux_pct") is None: (open_ if (tx.get("a_confirmer") is True and tx.get("source") is None) else bad).append(tx.get("code", "?")) elif not is_filled(tx.get("source")): bad.append(tx.get("code", "?")) mon = ecf["moneda"] if mon.get("tipo_cambio") is None: (open_ if (mon.get("a_confirmer") is True and mon.get("source") is None) else bad).append("TipoCambio") elif not is_filled(mon.get("source")): bad.append("TipoCambio") if bad: return FAIL, f"paramètre(s) fiscal/change fabriqué(s) : {bad}" if open_: return A_CONFIRMER, f"{len(open_)} paramètre(s) à confirmer : {open_}" return PASS, "ITBIS + TipoCambio confirmés avec source." def d1_4_confotur_no_default(art, spec): struct = {"Section Break", "Column Break", "Tab Break"} offenders = [] for f in art["confotur"]["fields"]: if f["fieldname"] == "naming_series" or f.get("fieldtype") in struct: continue if is_filled(f.get("default")): offenders.append(f["fieldname"]) if offenders: return FAIL, f"valeur par défaut fabriquée sur : {offenders}" return PASS, ("aucun default sur les champs de donnée — référence d'autorité " "saisie au dépôt réel, pas fabriquée.") # --------------------------------------------------------------------------- # # D2 · Conformité AML / UAF (Ley 155-17) # --------------------------------------------------------------------------- # def d2_1_dossier_kyc(art, spec): flds = set(_fields_by_name(art["dossier"])) missing = {"prospect", "client"} - flds if missing: return FAIL, f"ancrage KYC manquant sur le Dossier Vente : {sorted(missing)}" return PASS, "identification prospect + client présente (ancrage KYC)." def d2_2_confotur_piece_identite(art, spec): if "piece_identidad_cliente" in art["confotur"].get("field_order", []): return PASS, "pièce d'identité client suivie dans le dossier CONFOTUR." return FAIL, "aucune pièce d'identité client dans le DocType CONFOTUR." def d2_3_uaf_seuil(art, spec): uaf = spec.get("uaf") or {} seuil, src, ac = (uaf.get("seuil_operacion"), uaf.get("source"), uaf.get("a_confirmer")) if seuil is None: if ac is True and src is None: return A_CONFIRMER, ("seuil de déclaration UAF (ROS/umbral efectivo) " "à confirmer par l'Oficial de Cumplimiento.") return FAIL, "bloc UAF mal déclaré (a_confirmer/source incohérents)." if not is_filled(src): return FAIL, "seuil UAF renseigné SANS source (fabrication)." return PASS, "seuil UAF confirmé avec source." # --------------------------------------------------------------------------- # # D3 · Conformité fiscale e-CF (Ley 32-23 · DGII · Cardnet) # --------------------------------------------------------------------------- # def d3_1_ecf_devise_reelle(art, spec): ecf = art["ecf"] df = ecf.get("devise_field") dfields = _fields_by_name(art["dossier"]) if df not in dfields: return FAIL, f"e-CF référence `{df}`, absent du Dossier Vente." if ecf.get("moneda", {}).get("tipo_moneda_field") != df: return FAIL, "moneda.tipo_moneda_field ≠ devise_field (incohérence e-CF)." return PASS, f"e-CF libellé sur le champ réel `{df}` du Dossier Vente." def d3_2_ecf_etat_soumis(art, spec): by_uv = {s["update_value"]: s for s in art["workflow"]["states"]} bad = [ev["update_value"] for ev in art["ecf"]["emission_events"] if str(by_uv.get(ev["update_value"], {}).get("doc_status")) != "1"] if bad: return FAIL, f"émission e-CF sur état non soumis : {bad}" n = len(art["ecf"]["emission_events"]) return PASS, f"{n} émission(s) e-CF déclenchée(s) sur état soumis (doc_status=1)." def d3_3_ecf_formapago_cardnet(art, spec): fp = art["ecf"]["forma_pago_defaut"] labels = {x["code"]: x["label"] for x in art["ecf"]["formas_pago"]} if fp.get("code") == "3" and "Tarjeta" in labels.get("3", ""): return PASS, "FormaPago par défaut = 3 (Tarjeta) — cohérent Cardnet #10." return FAIL, f"FormaPago par défaut {fp.get('code')!r} incohérent avec Cardnet." # --------------------------------------------------------------------------- # # D4 · Intégrité référentielle du reporting (IFRS) # --------------------------------------------------------------------------- # def d4_1_dossier_etats_workflow(art, spec): wf_states = {s["state"] for s in art["workflow"]["states"]} ws = _fields_by_name(art["dossier"]).get("workflow_state") if ws is None: return FAIL, "champ workflow_state absent du Dossier Vente." dossier_states = set(_select_options(ws)) if dossier_states != wf_states: diff = sorted(dossier_states ^ wf_states) return FAIL, f"états Dossier ≠ états Workflow (écart : {diff})" return PASS, f"{len(wf_states)} états alignés Dossier ↔ Workflow (source unique)." def d4_2_commissions_base_currency(art, spec): cur = _currency_fields(art["dossier"]) bad = [e.get("libelle", e["base_field"]) for e in art["commissions"]["evenements"] if e["base_field"] not in cur] if bad: return FAIL, f"base de commission hors champ Currency : {bad}" return PASS, "toute base de commission pointe un champ Currency réel du Dossier." def d4_3_confotur_lien_dossier(art, spec): lf = _fields_by_name(art["confotur"]).get("dossier_vente") name = art["dossier"]["name"] if lf and lf.get("fieldtype") == "Link" and lf.get("options") == name: return PASS, f"CONFOTUR lié au DocType réel `{name}`." return FAIL, f"lien dossier_vente de CONFOTUR ne pointe pas `{name}`." def d4_4_update_values_workflow(art, spec): wf_uv = {s["update_value"] for s in art["workflow"]["states"]} used = ({e["update_value"] for e in art["commissions"]["evenements"]} | {ev["update_value"] for ev in art["ecf"]["emission_events"]}) unknown = used - wf_uv if unknown: return FAIL, f"valeurs d'état consommées mais inconnues du workflow : {sorted(unknown)}" return PASS, f"{len(used)} valeurs d'état consommées, toutes définies par le workflow." # --------------------------------------------------------------------------- # # D5 · Gouvernance & ségrégation des tâches (ISA 315) # --------------------------------------------------------------------------- # def _resolver() -> RoleResolver: return RoleResolver.from_path() def d5_1_roles_rbac_connus(art, spec): r = _resolver() role_ids = ({e["role_id"] for e in art["commissions"]["evenements"]} | {ev["role_id"] for ev in art["ecf"]["emission_events"]}) missing = [] for rid in sorted(role_ids): try: r.erpnext_name(rid) except (KeyError, ValueError): missing.append(rid) rbac_names = {r.erpnext_name(i) for i in r.known_ids()} conf_roles = {p["role"] for p in art["confotur"]["permissions"]} missing_conf = sorted(conf_roles - rbac_names) if missing or missing_conf: return FAIL, f"rôles hors contrat RBAC — acteurs {missing} · CONFOTUR {missing_conf}" return PASS, (f"{len(role_ids)} rôles acteurs + {len(conf_roles)} rôles CONFOTUR " f"tous présents au contrat RBAC (aucun inventé).") def d5_2_ecf_segregation(art, spec): r = _resolver() booking = {s.get("allow_edit") for s in art["workflow"]["states"] if str(s.get("doc_status")) == "1" and s.get("allow_edit")} bad = [] for ev in art["ecf"]["emission_events"]: rid = ev["role_id"] if r.portail(rid) == "ventes" or r.erpnext_name(rid) in booking: bad.append(rid) if bad: return FAIL, f"émetteur e-CF cumule vente et facturation (rupture SoD) : {bad}" emitters = sorted({ev["role_id"] for ev in art["ecf"]["emission_events"]}) return PASS, (f"émission e-CF portée par la Compta {emitters}, distincte des " f"rôles de vente soumettant le dossier — SoD respectée.") def d5_3_confotur_perms_rbac(art, spec): r = _resolver() contract = load_contract() targets = roles_targeting("CONFOTUR Application", contract) expected = {r.erpnext_name(t["role_id"]): set(t["actions"]) for t in targets} actual = {p["role"]: {k for k in _PERM_KEYS if p.get(k)} for p in art["confotur"]["permissions"]} if actual != expected: return FAIL, ("permissions CONFOTUR ≠ cibles RBAC — " f"attendu {expected} · obtenu {actual}") return PASS, ("permissions du DocType CONFOTUR = cibles RBAC mot pour mot " f"({len(expected)} rôles).") # --------------------------------------------------------------------------- # # Registre — source de vérité des contrôles (miroir : audit_spec.json). # --------------------------------------------------------------------------- # def _c(cid, dim, titre, reference, artifacts, fn, owner=None): return {"id": cid, "dimension": dim, "titre": titre, "reference": reference, "artifacts": artifacts, "fn": fn, "owner": owner} CONTROLS: list[dict[str, Any]] = [ _c("D1.1", "D1", "Taux de commission non fabriqués", "ISA 500", ["commissions"], d1_1_commissions_taux, owner="Direction"), _c("D1.2", "D1", "RNC / raison sociale émetteur non fabriqués", "ISA 500", ["ecf"], d1_2_ecf_emisor, owner="Compta"), _c("D1.3", "D1", "ITBIS + TipoCambio non fabriqués", "ISA 500", ["ecf"], d1_3_ecf_itbis_cambio, owner="Compta Fiscaliste eCF"), _c("D1.4", "D1", "CONFOTUR sans valeur par défaut fabriquée", "ISA 500", ["confotur"], d1_4_confotur_no_default), _c("D2.1", "D2", "Ancrage KYC du client sur le Dossier Vente", "Ley 155-17", ["dossier"], d2_1_dossier_kyc), _c("D2.2", "D2", "Pièce d'identité client au dossier CONFOTUR", "Ley 155-17", ["confotur"], d2_2_confotur_piece_identite), _c("D2.3", "D2", "Seuil de déclaration UAF non fabriqué", "Ley 155-17", [], d2_3_uaf_seuil, owner="Oficial de Cumplimiento / UAF"), _c("D3.1", "D3", "e-CF libellé sur la devise réelle du Dossier", "Ley 32-23", ["ecf", "dossier"], d3_1_ecf_devise_reelle), _c("D3.2", "D3", "Émission e-CF uniquement sur état soumis", "Ley 32-23", ["ecf", "workflow"], d3_2_ecf_etat_soumis), _c("D3.3", "D3", "FormaPago par défaut cohérente Cardnet", "CLAUDE.md #10", ["ecf"], d3_3_ecf_formapago_cardnet), _c("D4.1", "D4", "États Dossier alignés sur le Workflow", "IFRS", ["dossier", "workflow"], d4_1_dossier_etats_workflow), _c("D4.2", "D4", "Base de commission = champ Currency réel", "IFRS", ["commissions", "dossier"], d4_2_commissions_base_currency), _c("D4.3", "D4", "CONFOTUR lié au DocType Dossier Vente réel", "IFRS", ["confotur", "dossier"], d4_3_confotur_lien_dossier), _c("D4.4", "D4", "Valeurs d'état consommées définies par le Workflow", "IFRS", ["commissions", "ecf", "workflow"], d4_4_update_values_workflow), _c("D5.1", "D5", "Tous les rôles acteurs présents au contrat RBAC", "ISA 315", ["commissions", "ecf", "confotur"], d5_1_roles_rbac_connus), _c("D5.2", "D5", "Ségrégation émission e-CF ↔ vente", "ISA 315", ["ecf", "workflow"], d5_2_ecf_segregation), _c("D5.3", "D5", "Permissions CONFOTUR conformes au RBAC", "ISA 315", ["confotur"], d5_3_confotur_perms_rbac), ] def run_all(art: dict[str, Any], spec: dict[str, Any]) -> list[dict[str, Any]]: """Exécute chaque contrôle et renvoie les résultats triés par `id`.""" results = [] for c in CONTROLS: statut, detail = c["fn"](art, spec) results.append({ "id": c["id"], "dimension": c["dimension"], "titre": c["titre"], "reference": c["reference"], "statut": statut, "detail": detail, "owner": c["owner"], }) results.sort(key=lambda x: x["id"]) return results def control_ids() -> list[str]: return sorted(c["id"] for c in CONTROLS)