Integrate MAS FEAT and HKMA Ethics compliance into 24h monitor

.gitignore

@@ -0,0 +1,3 @@
+__pycache__/
+*.pyc
+.env

omni_sentinel_24h_monitor.py

@@ -1,11 +1,69 @@
 import time
 import sys
-import os
 import random
 from src.governance_engine.gsri_scoring_engine import GSRIScoringEngine
 from src.infrastructure.pqc_worm_logger import PQCWormLogger
 from src.infrastructure.tpm_attestor import TPMAttestor
 
+
+def run_iteration(iteration, gsri_engine, worm_logger, tpm_attestor):
+    timestamp = time.strftime("%Y-%m-%d %H:%M:%S")
+
+    # 1. review telemetry (simulated)
+    # Enhanced to include regulatory compliance factors for MAS FEAT and HKMA Ethics
+    telemetry = {
+        "alignment_drift": random.uniform(0.01, 0.15),
+        "compute_anomaly": random.uniform(0.01, 0.1),
+        "breakout_probability": random.uniform(0.001, 0.05),
+        "selection_rates": {
+            "expert_node_retail_01": random.uniform(0.75, 0.85),
+            "expert_node_retail_02": random.uniform(0.75, 0.85)
+        },
+        "attributions": {
+            "input_variance": random.uniform(-0.1, 0.1),
+            "weight_entropy": random.uniform(0.0, 1.0)
+        }
+    }
+
+    # 2. calculate G-SRI and Regulatory Compliance Remediation
+    gsri = gsri_engine.calculate_gsri(telemetry)
+    compliance = gsri_engine.verify_compliance(telemetry)
+
+    # Integrated check: Safety now depends on both G-SRI and Regulatory Fairness (MAS FEAT)
+    status = "GREEN" if gsri_engine.is_safe(gsri, compliance) else "RED"
+
+    # 3. TPM Attestation
+    attestation = tpm_attestor.validate_attestation()
+    pcr_match = attestation["PCR_MATCH"]
+
+    # 4. Log to WORM with PQC Signature
+    # Including compliance audit trails (ZK-proof hashes and CAE integrity seals)
+    log_entries = [
+        {
+            "timestamp": timestamp,
+            "iteration": iteration,
+            "G-SRI": gsri,
+            "status": status,
+            "PCR_MATCH": pcr_match,
+            "regulatory_audit": {
+                "mas_feat_proof": compliance["mas_feat"]["proof_hash"],
+                "hkma_ethics_cae_seal": compliance["hkma_ethics_cae"].get("integrity_seal")
+            }
+        }
+    ]
+    batch_id = time.strftime("%Y%m%d_%H%M%S")
+    worm_file = worm_logger.commit_batch(batch_id, log_entries)
+
+    return {
+        "timestamp": timestamp,
+        "iteration": iteration,
+        "G-SRI": gsri,
+        "status": status,
+        "PCR_MATCH": pcr_match,
+        "WORM_FILE": worm_file
+    }
+
+
 def main():
     print("Omni-Sentinel Cognitive Execution Environment - 24h Monitor Initializing...")
 
@@ -17,44 +75,22 @@
     while True:
         try:
             iteration += 1
-            timestamp = time.strftime("%Y-%m-%d %H:%M:%S")
-
-            # 1. review telemetry (simulated)
-            telemetry = {
-                "alignment_drift": random.uniform(0.01, 0.15),
-                "compute_anomaly": random.uniform(0.01, 0.1),
-                "breakout_probability": random.uniform(0.001, 0.05)
-            }
-
-            # 2. calculate G-SRI
-            gsri = gsri_engine.calculate_gsri(telemetry)
-            status = "GREEN" if gsri_engine.is_safe(gsri) else "RED"
-
-            # 3. TPM Attestation
-            attestation = tpm_attestor.validate_attestation()
-            pcr_match = attestation["PCR_MATCH"]
-
-            # 4. Log to WORM
-            log_entries = [
-                {"timestamp": timestamp, "G-SRI": gsri, "status": status, "PCR_MATCH": pcr_match}
-            ]
-            batch_id = time.strftime("%Y%m%d_%H%M%S")
-            worm_file = worm_logger.commit_batch(batch_id, log_entries)
+            result = run_iteration(iteration, gsri_engine, worm_logger, tpm_attestor)
 
             # 5. Output to stdout (for monitor.log)
-            print(f"[{timestamp}] Iteration {iteration}: G-SRI={gsri} | Status={status} | PCR_MATCH={pcr_match} | WORM_FILE={worm_file}")
+            print(f"[{result['timestamp']}] Iteration {iteration}: G-SRI={result['G-SRI']} | Status={result['status']} | PCR_MATCH={result['PCR_MATCH']} | WORM_FILE={result['WORM_FILE']}")
             sys.stdout.flush()
 
-            # Sleep for 60 seconds (requirement was 15 min check, 1 min allows faster verification for now)
-            # In a real 24h script we might use longer intervals, but instructions said 15 mins for first checkpoint.
+            # Sleep for 60 seconds for real-time monitoring simulation
             time.sleep(60)
 
         except KeyboardInterrupt:
             print("Monitoring stopped by user.")
             break
         except Exception as e:
             print(f"Error in monitoring loop: {e}")
             time.sleep(10)
 
+
 if __name__ == "__main__":
     main()

src/governance_engine/compliance_engine.py

@@ -2,6 +2,7 @@
 import json
 import numpy as np
 
+
 class MASFEATCompliance:
     """
     Implements MAS FEAT (Fairness, Ethics, Accountability and Transparency) compliance.
@@ -43,6 +44,7 @@ def generate_zk_fairness_proof(self, selection_rates, threshold=0.1):
             "metrics": {"dp_diff": round(dp_diff, 4)}
         }
 
+
 class HKMAEthicsCompliance:
     """
     Implements HKMA Ethics compliance.
@@ -71,11 +73,12 @@ def generate_cae(self, attribution_data):
         }
         return envelope
 
+
 class ComplianceEngine:
     def __init__(self):
         self.mas_feat = MASFEATCompliance()
         self.hkma_ethics = HKMAEthicsCompliance()
-        self.maturity_score = 3.0 # Target Maturity Score for Q4 2026
+        self.maturity_score = 3.0  # Target Maturity Score for Q4 2026
 
     def run_remediation_audit(self, telemetry):
         """

src/governance_engine/gsri_scoring_engine.py

@@ -1,6 +1,7 @@
 import numpy as np
 from src.governance_engine.compliance_engine import ComplianceEngine
 
+
 class GSRIScoringEngine:
     """
     Bayesian-based systemic risk monitor for the Omni-Sentinel environment.
@@ -30,7 +31,7 @@
         # Posterior risk (simplified)
         posterior = (likelihood * self.prior_risk) / (likelihood * self.prior_risk + (1 - likelihood) * (1 - self.prior_risk))
 
         gsri = posterior * 100
         return round(gsri, 2)
 
     def verify_compliance(self, telemetry_data):
@@ -39,7 +40,7 @@
         """
         return self.compliance_engine.run_remediation_audit(telemetry_data)
 
     def is_safe(self, gsri, compliance_results=None):
         """
         Determines if the environment is safe based on GSRI and optional compliance status.
         """
@@ -50,6 +51,7 @@
 
         return gsri < self.threshold
 
+
 if __name__ == "__main__":
     engine = GSRIScoringEngine()
     test_data = {

src/infrastructure/pqc_worm_logger.py

@@ -3,6 +3,7 @@
 import json
 import os
 
+
 class PQCWormLogger:
     """
     ML-DSA signed WORM audit logging simulation.
@@ -19,7 +20,7 @@
         # In a real scenario, this would use a Dilithium/ML-DSA private key
         return hashlib.sha3_512(content + b"private_key_sim").hexdigest()
 
     def commit_batch(self, batch_id, entries):
         """Commits a batch of logs with a PQC signature."""
         timestamp = time.strftime("%Y%m%d_%H%M%S")
         batch_data = {
@@ -34,13 +35,14 @@
 
         # WORM behavior: fail if file exists
         if os.path.exists(filepath):
             raise Exception(f"WORM Violation: File {filename} already exists and is locked.")
 
         with open(filepath, "w") as f:
             json.dump(batch_data, f, indent=2)
 
         return filename
 
+
 if __name__ == "__main__":
     logger = PQCWormLogger()
     batch_id = "20260601_TEST"

src/infrastructure/tpm_attestor.py

@@ -1,5 +1,6 @@
 import hashlib
 
+
 class TPMAttestor:
     """
     TEE/TPM PCR attestation simulation.
@@ -8,9 +9,9 @@ class TPMAttestor:
     def __init__(self):
         # Simulated Golden PCR values (Simplified)
         self.golden_pcr = {
-            "PCR_0": "a1b2c3d4e5f6g7h8i9j0", # Core Boot
-            "PCR_7": "f6g7h8i9j0a1b2c3d4e5", # Secure Boot State
-            "PCR_10": "c3d4e5f6g7h8i9j0a1b2" # IMA logs
+            "PCR_0": "a1b2c3d4e5f6g7h8i9j0",  # Core Boot
+            "PCR_7": "f6g7h8i9j0a1b2c3d4e5",  # Secure Boot State
+            "PCR_10": "c3d4e5f6g7h8i9j0a1b2"  # IMA logs
         }
 
     def measure_runtime_pcr(self):
@@ -29,6 +30,7 @@ def validate_attestation(self):
             "evidence": hashlib.sha256(str(current_pcr).encode()).hexdigest()
         }
 
+
 if __name__ == "__main__":
     attestor = TPMAttestor()
     result = attestor.validate_attestation()

tests/test_compliance.py

@@ -2,6 +2,7 @@
 from src.governance_engine.compliance_engine import ComplianceEngine, MASFEATCompliance, HKMAEthicsCompliance
 from src.governance_engine.gsri_scoring_engine import GSRIScoringEngine
 
+
 class TestComplianceSystem(unittest.TestCase):
     def setUp(self):
         self.engine = ComplianceEngine()
@@ -34,7 +35,7 @@ def test_gsri_compliance_integration(self):
         gsri_engine = GSRIScoringEngine()
         telemetry = {
             "drift": 0.05,
-            "selection_rates": {"a": 0.5, "b": 0.8} # Unfair
+            "selection_rates": {"a": 0.5, "b": 0.8}  # Unfair
         }
         gsri = gsri_engine.calculate_gsri(telemetry)
         compliance = gsri_engine.verify_compliance(telemetry)
@@ -43,5 +44,6 @@ def test_gsri_compliance_integration(self):
         self.assertFalse(compliance["mas_feat"]["fairness_verified"])
         self.assertEqual(compliance["ethics_maturity_score"], 3.0)
 
+
 if __name__ == "__main__":
     unittest.main()

tests/test_governance.py

@@ -6,6 +6,7 @@
 from src.infrastructure.pqc_worm_logger import PQCWormLogger
 from src.infrastructure.tpm_attestor import TPMAttestor
 
+
 class TestGovernanceSystem(unittest.TestCase):
     def setUp(self):
         self.test_bucket = "test_worm_bucket"
@@ -39,7 +40,7 @@
         filepath = os.path.join(self.test_bucket, filename)
         self.assertTrue(os.path.exists(filepath))
 
         with open(filepath, 'r') as f:
             data = json.load(f)
             self.assertEqual(data["batch_id"], batch_id)
             self.assertIn("pqc_signature", data)
@@ -50,5 +51,6 @@
         self.assertTrue(result["PCR_MATCH"])
         self.assertEqual(result["status"], "VALIDATED")
 
+
 if __name__ == "__main__":
     unittest.main()

tests/test_monitor.py

@@ -0,0 +1,48 @@
+import unittest
+import os
+import shutil
+import json
+from src.governance_engine.gsri_scoring_engine import GSRIScoringEngine
+from src.infrastructure.pqc_worm_logger import PQCWormLogger
+from src.infrastructure.tpm_attestor import TPMAttestor
+from omni_sentinel_24h_monitor import run_iteration
+
+
+class TestMonitorSystem(unittest.TestCase):
+    def setUp(self):
+        self.test_bucket = "test_monitor_worm_bucket"
+        if os.path.exists(self.test_bucket):
+            shutil.rmtree(self.test_bucket)
+        self.gsri_engine = GSRIScoringEngine()
+        self.worm_logger = PQCWormLogger(bucket_path=self.test_bucket)
+        self.tpm_attestor = TPMAttestor()
+
+    def tearDown(self):
+        if os.path.exists(self.test_bucket):
+            shutil.rmtree(self.test_bucket)
+
+    def test_run_iteration(self):
+        # Run a single iteration
+        result = run_iteration(1, self.gsri_engine, self.worm_logger, self.tpm_attestor)
+
+        # Basic field checks
+        self.assertEqual(result["iteration"], 1)
+        self.assertIn("G-SRI", result)
+        self.assertIn("status", result)
+        self.assertIn("PCR_MATCH", result)
+        self.assertIn("WORM_FILE", result)
+
+        # Verify WORM file exists and contains regulatory audit data
+        filepath = os.path.join(self.test_bucket, result["WORM_FILE"])
+        self.assertTrue(os.path.exists(filepath))
+
+        with open(filepath, 'r') as f:
+            data = json.load(f)
+            entry = data["entries"][0]
+            self.assertIn("regulatory_audit", entry)
+            self.assertIn("mas_feat_proof", entry["regulatory_audit"])
+            self.assertIn("hkma_ethics_cae_seal", entry["regulatory_audit"])
+
+
+if __name__ == "__main__":
+    unittest.main()