#!/usr/bin/env python3 """MICO-JDEQ Identity Awareness Engine M2M-001 HARDENED""" import json, os, re, sys, uuid, platform, hashlib from datetime import datetime, timezone from pathlib import Path from typing import Any, Dict, List, Optional, Tuple __version__ = "0.1.0-hardened" MICO_ROOT = Path(os.getenv("MICO_JDEQ_ROOT", "/mico")) IDENTITY_DIR = MICO_ROOT / "02_CAPABILITY" / "identity" CONFIG_DIR = IDENTITY_DIR / "config" DEFAULT_CONFIG_PATH = CONFIG_DIR / "identity.json" DEFAULT_CONTRACT_PATH = MICO_ROOT / "01_GOVERNANCE" / "contract.md" DEFAULT_EVIDENCE_DIR = MICO_ROOT / "05_EVIDENCE" FIELD_SCORES = { "system_name": 10, "architecture_version": 10, "node_id": 15, "node_role": 15, "owner": 10, "ssot_revision": 10, "policy_revision": 10, "runtime_version": 10, "boot_timestamp": 10, "contract_hash": 10 } TOTAL_POINTS = sum(FIELD_SCORES.values()) VALID_NODE_ROLES = { "CONTROL_CENTER", "HEAVY_WORKER", "EDGE_INTERFACE", "EDGE_BACKUP", "ELASTIC_POOL" } def validate_system_name(v): return isinstance(v, str) and len(v.strip()) > 0 def validate_architecture_version(v): return isinstance(v, str) and bool(re.match(r"^M\d+M-\d+\.\d+$", v)) def validate_node_id(v): return isinstance(v, str) and len(v) >= 3 and bool(re.match(r"^[A-Za-z0-9\-_]+$", v)) def validate_node_role(v): return isinstance(v, str) and v in VALID_NODE_ROLES def validate_owner(v): return isinstance(v, str) and len(v.strip()) > 0 def validate_revision(v): return isinstance(v, str) and bool(re.match(r"^[A-Z0-9]+-\d+$", v)) def validate_runtime_version(v): return isinstance(v, str) and bool(re.match(r"^\d+\.\d+\.\d+", v)) def validate_iso8601_timestamp(v): if not isinstance(v, str): return False try: datetime.fromisoformat(v.replace("Z", "+00:00")) return True except ValueError: return False def validate_sha256_hash(v): return isinstance(v, str) and bool(re.match(r"^[a-f0-9]{64}$", v)) VALIDATORS = { "system_name": validate_system_name, "architecture_version": validate_architecture_version, "node_id": validate_node_id, "node_role": validate_node_role, "owner": validate_owner, "ssot_revision": validate_revision, "policy_revision": validate_revision, "runtime_version": validate_runtime_version, "boot_timestamp": validate_iso8601_timestamp, "contract_hash": validate_sha256_hash } def _sha256(data: str) -> str: return hashlib.sha256(data.encode("utf-8")).hexdigest() def _load_config(path: Path) -> Tuple[Dict[str, Any], Optional[str], Optional[str]]: config_hash = None try: with path.open("r", encoding="utf-8") as f: content = f.read() config = json.loads(content) config_hash = _sha256(content) return config, None, config_hash except FileNotFoundError: return {}, f"Config file not found: {path}", None except PermissionError: return {}, f"Permission denied: {path}", None except json.JSONDecodeError as e: return {}, f"Invalid JSON: {e}", None def _load_contract_hash(path: Path) -> Optional[str]: if path.exists(): try: return _sha256(path.read_text(encoding="utf-8")) except Exception: return None return None def _get_boot_timestamp() -> Optional[str]: try: with open("/proc/uptime", "r") as f: uptime = float(f.readline().split()[0]) boot = datetime.now(timezone.utc).timestamp() - uptime return datetime.fromtimestamp(boot, tz=timezone.utc).isoformat() except Exception: return None def _get_runtime_data() -> Dict[str, Any]: return { "runtime_version": platform.python_version(), "system_name": "MICO-JDEQ", "node_id": platform.node() or "unknown", "boot_timestamp": _get_boot_timestamp(), } class IdentityEngine: def __init__(self, config_path=None, contract_path=None, evidence_dir=None): self.config_path = config_path or DEFAULT_CONFIG_PATH self.contract_path = contract_path or DEFAULT_CONTRACT_PATH self.evidence_dir = evidence_dir or DEFAULT_EVIDENCE_DIR self.identity = {} self.missing_fields = [] self.score_breakdown = {} self.validity_score = 0 self.hard_gate = {} self.status = "" self.error_message = None self.config_hash = None self.contract_hash = None def collect(self): config, err, config_hash = _load_config(self.config_path) self.error_message = err self.config_hash = config_hash runtime = _get_runtime_data() self.identity = {**runtime, **config} self.contract_hash = _load_contract_hash(self.contract_path) self.identity["contract_hash"] = self.contract_hash def validate(self): self.missing_fields = [] self.score_breakdown = {} total_points = 0 for field, validator in VALIDATORS.items(): value = self.identity.get(field) if validator(value): points = FIELD_SCORES.get(field, 0) self.score_breakdown[field] = points total_points += points else: self.score_breakdown[field] = 0 self.missing_fields.append(field) self.validity_score = round((total_points / TOTAL_POINTS) * 100) if TOTAL_POINTS else 0 system_name_present = validate_system_name(self.identity.get("system_name")) node_id_valid = validate_node_id(self.identity.get("node_id")) node_role_valid = validate_node_role(self.identity.get("node_role")) gate_passed = system_name_present and node_id_valid and node_role_valid self.hard_gate = { "node_id_valid": node_id_valid, "node_role_valid": node_role_valid, "system_name_present": system_name_present, "gate_passed": gate_passed, "gate_version": "1.0", "policy_version": os.getenv("MICO_POLICY_VERSION", "POL-001"), "gate_evaluation_timestamp": datetime.now(timezone.utc).isoformat(), } if not gate_passed: self.status = "NOT_VERIFIED" elif self.validity_score >= 90: self.status = "VERIFIED" elif self.validity_score >= 70: self.status = "PARTIAL" elif self.validity_score >= 40: self.status = "WEAK" else: self.status = "INVALID" def generate_output(self): identity_canonical = json.dumps(dict(sorted(self.identity.items())), sort_keys=True, default=str) identity_hash = _sha256(identity_canonical) trace_id = uuid.uuid4().hex evidence = { "data_source": "local_system + config_file + contract_file", "collection_method": "runtime_probe", "config_hash": self.config_hash, "contract_hash": self.contract_hash, "identity_hash": identity_hash, "evidence_hash": "", "trace_id": trace_id, } output = { "module": "identity_awareness_engine", "version": __version__, "timestamp": datetime.now(timezone.utc).isoformat(), "identity": self.identity, "score_breakdown": self.score_breakdown, "validity_score": self.validity_score, "hard_gate": self.hard_gate, "status": self.status, "missing_fields": self.missing_fields, "evidence": evidence, } if self.error_message: output["error"] = self.error_message output_copy = json.loads(json.dumps(output, sort_keys=True, default=str)) output_copy["evidence"]["evidence_hash"] = "" output["evidence"]["evidence_hash"] = _sha256(json.dumps(output_copy, sort_keys=True, default=str)) return output def save_output(self, output): try: self.evidence_dir.mkdir(parents=True, exist_ok=True) out_path = self.evidence_dir / "identity_state.json" with open(out_path, "w", encoding="utf-8") as f: json.dump(output, f, indent=2, default=str, sort_keys=True) except Exception as e: print(f"[ERROR] Gagal menyimpan evidence: {e}", file=sys.stderr) def run(self): self.collect() self.validate() output = self.generate_output() self.save_output(output) return output def main(): engine = IdentityEngine() output = engine.run() print(json.dumps(output, indent=2, default=str, sort_keys=True)) if output["status"] == "VERIFIED": return 0 elif output["status"] in ("PARTIAL", "WEAK"): return 1 else: return 2 if __name__ == "__main__": sys.exit(main())