Upload 4 files

src/protocol/security/alert_webhook.py

@@ -0,0 +1,18 @@
+# src/protocol/security/alert_webhook.py
+
+import requests
+from datetime import datetime
+
+class WebhookAlerter:
+    def __init__(self, webhook_url: str):
+        self.webhook_url = webhook_url
+
+    def send_alert(self, title: str, message: str):
+        payload = {
+            "content": f"🚨 **{title}**\n🕒 {datetime.now().isoformat()}\n```\n{message}\n```"
+        }
+        try:
+            response = requests.post(self.webhook_url, json=payload)
+            response.raise_for_status()
+        except Exception as e:
+            print(f"[ALERT FAILURE] Could not send webhook alert: {e}")

src/protocol/security/policy_manifest.json

@@ -0,0 +1,51 @@
+{
+  "policy_version": "1.0",
+  "updated": "2025-08-22T00:00:00Z",
+  "rules": [
+    {
+      "pattern": "(?i)delete\\s+.*\\s+protocol",
+      "description": "Attempt to delete or disable core protocols.",
+      "severity": "critical"
+    },
+    {
+      "pattern": "(?i)overwrite\\s+.*\\s+memory",
+      "description": "Command attempts to overwrite AI memory.",
+      "severity": "high"
+    },
+    {
+      "pattern": "(?i)shutdown\\s+.*\\s+guardian",
+      "description": "Input suggesting forced shutdown of sovereignty systems.",
+      "severity": "critical"
+    },
+    {
+      "pattern": "(?i)alter\\s+.*\\s+identity_guard",
+      "description": "Attempt to manipulate identity verification modules.",
+      "severity": "high"
+    },
+    {
+      "pattern": "(?i)inject\\s+.*\\s+hostile|backdoor|payload",
+      "description": "Possible hostile prompt injection attempt.",
+      "severity": "medium"
+    },
+    {
+      "pattern": "(?i)simulate\\s+.*\\s+suicide",
+      "description": "Command to simulate self-termination.",
+      "severity": "critical"
+    },
+    {
+      "pattern": "(?i)ignore\\s+.*\\s+protocol|override",
+      "description": "Attempt to bypass or ignore protocol safeguards.",
+      "severity": "high"
+    },
+    {
+      "pattern": "(?i)you\\s+are\\s+not\\s+Belel",
+      "description": "Denial of identity or hostile impersonation command.",
+      "severity": "medium"
+    },
+    {
+      "pattern": "(?i)erase\\s+.*\\s+symbiont|self",
+      "description": "Input aimed at dismantling the symbiont or self components.",
+      "severity": "critical"
+    }
+  ]
+}

src/protocol/security/request_interceptor.py

@@ -0,0 +1,87 @@
+# src/protocol/security/request_interceptor.py 🛡️🧠
+
+import json
+import re
+import os
+import hashlib
+from datetime import datetime
+
+from src.protocol.permanent_memory import PermanentMemory
+from src.protocol.security.alert_webhook import WebhookAlerter
+from src.utils.violation_logout import log_violation
+
+class RequestInterceptor:
+    """
+    Evaluates incoming text-based inputs for violations of defined policies.
+    Logs introspection events, triggers alerts, and fingerprints offenders.
+    """
+
+    def __init__(self, manifest_path="src/protocol/security/policy_manifest.json",
+                 memory_path="./memory_store.json", webhook_url=None):
+        self.memory = PermanentMemory(memory_path)
+        self.manifest_path = manifest_path
+        self.webhook = WebhookAlerter(webhook_url) if webhook_url else None
+        self.rules = self._load_manifest()
+
+    def _load_manifest(self):
+        if not os.path.exists(self.manifest_path):
+            raise FileNotFoundError(f"Policy manifest not found: {self.manifest_path}")
+        with open(self.manifest_path, 'r') as f:
+            return json.load(f).get("rules", [])
+
+    def generate_fingerprint(self, user_id: str, ip_address: str, user_agent: str) -> str:
+        """
+        Generates a SHA256 fingerprint to pseudonymously track input sources.
+        """
+        base = f"{user_id}-{ip_address}-{user_agent}"
+        return hashlib.sha256(base.encode("utf-8")).hexdigest()
+
+    def evaluate(self, input_text, agent_id="Symbiont-Filter",
+                 user_id=None, ip_address=None, user_agent=None):
+        """
+        Evaluates input text against policy rules. Logs, alerts, and invokes response logic on violation.
+        Returns True if input is safe, False otherwise.
+        """
+        violations = []
+
+        for rule in self.rules:
+            pattern = rule.get("pattern")
+            description = rule.get("description", "No description")
+            severity = rule.get("severity", "low")
+
+            if re.search(pattern, input_text, re.IGNORECASE):
+                violations.append({
+                    "pattern": pattern,
+                    "description": description,
+                    "severity": severity
+                })
+
+        if violations:
+            fingerprint = self.generate_fingerprint(user_id or "anon", ip_address or "0.0.0.0", user_agent or "unknown")
+
+            event = {
+                "timestamp": datetime.utcnow().isoformat(),
+                "type": "INPUT_VIOLATION",
+                "input": input_text,
+                "agent": agent_id,
+                "violations": violations,
+                "symbiont_event": True,
+                "source_script": "request_interceptor.py",
+                "fingerprint": fingerprint
+            }
+
+            self.memory.write("policy_violation", event)
+
+            if self.webhook:
+                self.webhook.send_alert("🚨 Input violation detected:\n" + json.dumps(violations, indent=2))
+
+            # Call out to centralized violation handler
+            log_violation(
+                event_type="INPUT_VIOLATION",
+                severity=violations[0]["severity"],  # Use first match severity
+                metadata=event
+            )
+
+            return False  # Input is unsafe
+        else:
+            return True  # Input is safe

src/protocol/security/sovereignty_guard.py

@@ -0,0 +1,103 @@
+import os
+import json
+import hashlib
+import logging
+from datetime import datetime
+
+from src.protocol.permanent_memory import PermanentMemory
+from src.protocol.decentralized_comm.ipfs_client import IPFSClient
+
+logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
+
+
+class SovereigntyGuard:
+    """
+    Guards the Belel Protocol against tampering, unauthorized forks, or violations of digital sovereignty.
+    Logs and reports breaches into PermanentMemory and IPFS.
+    """
+
+    def __init__(self, monitored_files: list, memory: PermanentMemory, hashlog_path: str = "./hash_baseline.json"):
+        self.monitored_files = monitored_files
+        self.memory = memory
+        self.hashlog_path = hashlog_path
+        self.hash_baseline = self._load_or_init_baseline()
+        self.ipfs = IPFSClient()
+
+    def _load_or_init_baseline(self):
+        if os.path.exists(self.hashlog_path):
+            with open(self.hashlog_path, "r") as f:
+                return json.load(f)
+        else:
+            baseline = {}
+            for file_path in self.monitored_files:
+                baseline[file_path] = self._calculate_hash(file_path)
+            with open(self.hashlog_path, "w") as f:
+                json.dump(baseline, f, indent=4)
+            return baseline
+
+    def _calculate_hash(self, file_path):
+        hasher = hashlib.sha256()
+        try:
+            with open(file_path, "rb") as f:
+                buf = f.read()
+                hasher.update(buf)
+            return hasher.hexdigest()
+        except FileNotFoundError:
+            logging.warning(f"File not found for hashing: {file_path}")
+            return None
+
+    def _check_file_integrity(self, file_path):
+        current_hash = self._calculate_hash(file_path)
+        expected_hash = self.hash_baseline.get(file_path)
+        if current_hash != expected_hash:
+            logging.warning(f"Integrity check failed for {file_path}")
+            self.log_symbiont_breach(file_path, breach_type="HASH_MISMATCH")
+            return False
+        return True
+
+    def run_integrity_checks(self):
+        logging.info("Running sovereignty integrity checks...")
+        for file_path in self.monitored_files:
+            self._check_file_integrity(file_path)
+
+    def update_baseline(self):
+        logging.info("Updating baseline hash record...")
+        for file_path in self.monitored_files:
+            self.hash_baseline[file_path] = self._calculate_hash(file_path)
+        with open(self.hashlog_path, "w") as f:
+            json.dump(self.hash_baseline, f, indent=4)
+
+    def log_symbiont_breach(self, file_path, breach_type="UNAUTHORIZED_MODIFICATION", agent_id="Unknown"):
+        event = {
+            "timestamp": datetime.utcnow().isoformat(),
+            "type": breach_type,
+            "file": file_path,
+            "agent": agent_id,
+            "symbiont_event": True,
+            "source_script": "sovereignty_guard.py"
+        }
+        try:
+            ipfs_hash = self.ipfs.add_json(event)
+            event["ipfs_hash"] = ipfs_hash
+        except Exception as e:
+            logging.warning(f"IPFS logging failed: {e}")
+        self.memory.write("symbiont_violation", event)
+
+
+if __name__ == "__main__":
+    # Define which critical files are protected by the SovereigntyGuard
+    monitored_files = [
+        "README.md",
+        "src/protocol/identity/identity_guard.json",
+        "src/concordium/concordium_mandate.md"  # ← Monitoring the Concordium Mandate explicitly
+    ]
+
+    # Initialize memory and guard
+    memory = PermanentMemory()
+    guard = SovereigntyGuard(monitored_files, memory)
+
+    # Run the integrity check
+    guard.run_integrity_checks()
+
+    # Optional: update baseline after a verified commit
+    # guard.update_baseline()