Create n8n_int_spec

n8n_int_spec

@@ -0,0 +1,560 @@
+EPISTEMIC THREAT MODEL & VALIDATOR ONTOLOGY
+
+FORMAL THREAT MODEL (STRIDE-E: Epistemic Extension)
+
+Spoofing - Identity Subversion
+
+```
+Threat: n8n workflow impersonates validator
+Impact: False authority injection into ledger
+Mitigation:
+1. Cryptographic validator identity (PKI-based)
+2. Validator role attestation signed by IRE
+3. Workflow-to-validator binding in ledger metadata
+
+Detection: Mismatch between workflow signature and validator claim
+Severity: Critical (sovereignty breach)
+```
+
+Tampering - Evidence Manipulation
+
+```
+Threat: n8n alters evidence pre-canonicalization
+Impact: Garbage-in, narrative-out
+Mitigation:
+1. Raw evidence fingerprinting (content hash before processing)
+2. Evidence lineage tracking in n8n workflow logs
+3. IRE detects fingerprint mismatches
+
+Detection: Pre-canonical hash ≠ post-canonical derivation
+Severity: Critical (truth contamination)
+```
+
+Repudiation - Epistemic Deniability
+
+```
+Threat: n8n denies triggering detection that found suppression
+Impact: System loses accountability for its own findings
+Mitigation:
+1. Non-repudiable workflow execution proofs
+2. Watermarked intermediate results
+3. Cross-referenced timestamp chains
+
+Detection: Missing execution proof for detection result
+Severity: High (accountability loss)
+```
+
+Information Disclosure - Pattern Leakage
+
+```
+Threat: Detection patterns leaked through n8n logs
+Impact: Adversaries learn system's detection thresholds
+Mitigation:
+1. Threshold abstraction (IRE returns categories, not scores)
+2. Differential privacy on aggregated results
+3. Ephemeral detection sessions
+
+Detection: Raw thresholds appear in orchestration logs
+Severity: Medium (detection model compromise)
+```
+
+Denial of Service - Epistemic Exhaustion
+
+```
+Threat: Flood system with nonsense to waste detection capacity
+Impact: Real patterns missed due to noise saturation
+Mitigation:
+1. Epistemic rate limiting per source
+2. Credibility-based throttling
+3. Detection priority queuing
+
+Detection: High-volume, low-signal detection requests
+Severity: Medium (resource exhaustion)
+```
+
+Elevation of Privilege - Ontology Hijacking
+
+```
+Threat: n8n attempts to modify lens/method definitions
+Impact: Epistemic framework compromise
+Mitigation:
+1. Immutable ontology registry (signed by originators)
+2. Versioned ontology with migration proofs
+3. No runtime ontology modification API
+
+Detection: Attempt to modify lens/method definitions
+Severity: Critical (sovereignty destruction)
+```
+
+Epistemic Drift - Gradual Corruption
+
+```
+Threat: Slow, subtle contamination of detection patterns
+Impact: System gradually aligns with preferred narrative
+Mitigation:
+1. Drift detection via historical self-comparison
+2. Cross-validation with frozen model versions
+3. Epistemic checksums on detection algorithms
+
+Detection: Statistical divergence from historical baseline
+Severity: High (stealth corruption)
+```
+
+VALIDATOR ONTOLOGY (Role-Based Authority)
+
+Validator Archetypes
+
+```
+1. Human-Sovereign Validator
+Role: Individual sovereignty preservation
+Authority: Can veto any ledger commit
+Identity: Self-sovereign cryptographic identity
+Quorum: Always required (cannot be automated away)
+Attestation: "I have reviewed and assert my sovereignty"
+
+2. System-Epistemic Validator  
+Role: Detection methodology integrity
+Authority: Validates detection process adherence
+Identity: Cryptographic hash of detection pipeline config
+Quorum: Required for automated commits
+Attestation: "Detection executed per published methodology"
+
+3. Source-Provenance Validator
+Role: Evidence chain custody
+Authority: Validates evidence hasn't been manipulated
+Identity: Hash of evidence handling workflow
+Quorum: Optional but recommended
+Attestation: "Evidence chain intact from source to canonicalization"
+
+4. Temporal-Integrity Validator
+Role: Time-bound execution verification
+Authority: Validates timestamps and execution windows
+Identity: Time-locked cryptographic proof
+Quorum: Required for time-sensitive commits
+Attestation: "Execution occurred within valid time window"
+
+5. Community-Plurality Validator
+Role: Cross-interpreter consensus
+Authority: Requires multiple human interpretations
+Identity: Set of interpreter identities + attestation
+Quorum: Variable based on interpretation count
+Attestation: "Multiple independent interpretations concur"
+```
+
+Validator Configuration Schema
+
+```json
+{
+  "validator_id": "urn:ire:validator:human_sovereign:sha256-abc123",
+  "archetype": "human_sovereign",
+  "authority_scope": ["ledger_commit", "evidence_rejection"],
+  "quorum_requirements": {
+    "minimum": 1,
+    "maximum": null,
+    "exclusivity": ["system_epistemic"]
+  },
+  "attestation_format": {
+    "required_fields": ["review_timestamp", "sovereignty_assertion"],
+    "signature_algorithm": "ed25519",
+    "expiration": "24h"
+  },
+  "epistemic_constraints": {
+    "cannot_override": ["detection_results", "canonical_hash"],
+    "can_reject_for": ["procedural_violation", "sovereignty_concern"]
+  }
+}
+```
+
+Validator Quorum Calculus
+
+```python
+def calculate_quorum_satisfaction(validators: List[Validator], commit_type: str) -> bool:
+    """Calculate if validator quorum is satisfied for commit type"""
+    
+    archetype_counts = Counter(v.archetype for v in validators)
+    
+    # Base requirements
+    requirements = {
+        "ledger_commit": {
+            "human_sovereign": 1,
+            "system_epistemic": 1,
+            "source_provenance": 0,  # optional
+            "temporal_integrity": 1,
+            "community_plurality": 0  # depends on interpretation count
+        },
+        "evidence_ingestion": {
+            "human_sovereign": 0,
+            "system_epistemic": 1,
+            "source_provenance": 1,
+            "temporal_integrity": 1,
+            "community_plurality": 0
+        },
+        "detection_escalation": {
+            "human_sovereign": 1,
+            "system_epistemic": 1,
+            "source_provenance": 0,
+            "temporal_integrity": 1,
+            "community_plurality": 1  # required for high-stakes escalations
+        }
+    }
+    
+    req = requirements[commit_type]
+    
+    # Check each archetype requirement
+    for archetype, required_count in req.items():
+        if archetype_counts.get(archetype, 0) < required_count:
+            return False
+    
+    # Check exclusivity constraints
+    for validator in validators:
+        for exclusive_archetype in validator.quorum_requirements.get("exclusivity", []):
+            if exclusive_archetype in archetype_counts:
+                return False
+    
+    return True
+```
+
+LEDGER SCHEMA HARDENING
+
+Extended Block Schema
+
+```json
+{
+  "block": {
+    "header": {
+      "id": "blk_timestamp_hash",
+      "prev": "previous_block_hash",
+      "timestamp": "ISO8601_with_nanoseconds",
+      "epistemic_epoch": 1,
+      "ontology_version": "sha256:ontology_hash"
+    },
+    "body": {
+      "nodes": [RealityNode],
+      "detection_context": {
+        "workflow_hash": "sha256:n8n_workflow_def",
+        "execution_window": {
+          "not_before": "timestamp",
+          "not_after": "timestamp",
+          "time_proof": "signature_from_temporal_validator"
+        },
+        "threshold_used": "abstract_category_not_numeric"
+      }
+    },
+    "validations": {
+      "attestations": [
+        {
+          "validator_id": "urn:ire:validator:...",
+          "archetype": "human_sovereign",
+          "attestation": "cryptographic_signature",
+          "scope": ["ledger_commit"],
+          "expires": "timestamp"
+        }
+      ],
+      "quorum_satisfied": true,
+      "quorum_calc": {
+        "required": {"human_sovereign": 1, "system_epistemic": 1},
+        "present": {"human_sovereign": 1, "system_epistemic": 1}
+      }
+    },
+    "integrity_marks": {
+      "evidence_fingerprint": "sha256_of_raw_content",
+      "detection_watermark": "nonce_based_on_workflow_id",
+      "epistemic_checksum": "hash_of_detection_logic_version"
+    }
+  }
+}
+```
+
+Detection Threshold Abstraction Layer
+
+```python
+class EpistemicThresholdInterface:
+    """Abstract threshold interface - n8n never sees numeric thresholds"""
+    
+    def __init__(self, ire_client):
+        self.ire = ire_client
+        
+    def should_escalate(self, detection_results: Dict) -> Dict:
+        """IRE decides escalation, returns abstract category"""
+        response = self.ire.post("/ire/detect/evaluate", {
+            "results": detection_results,
+            "return_format": "abstract_category"
+        })
+        
+        return {
+            "escalation_recommended": response.get("category") == "high_confidence",
+            "confidence_level": response.get("confidence_label"),  # "high"/"medium"/"low"
+            "next_action": response.get("recommended_action"),
+            # NO NUMERIC THRESHOLDS EXPOSED
+            # NO RAW SCORES EXPOSED
+        }
+    
+    def get_validation_requirements(self, category: str) -> Dict:
+        """Map escalation category to validator requirements"""
+        mapping = {
+            "high_confidence": {
+                "human_sovereign": 2,
+                "system_epistemic": 1,
+                "community_plurality": 1
+            },
+            "medium_confidence": {
+                "human_sovereign": 1,
+                "system_epistemic": 1
+            },
+            "low_confidence": {
+                "system_epistemic": 1
+            }
+        }
+        return mapping.get(category, {})
+```
+
+Time-Window Enforcement
+
+```python
+class TemporalIntegrityEnforcer:
+    """Enforce time-bound execution with cryptographic proofs"""
+    
+    def create_execution_window(self, duration_hours: int = 24) -> Dict:
+        """Create cryptographically bound execution window"""
+        window_id = f"window_{uuid.uuid4()}"
+        not_before = datetime.utcnow()
+        not_after = not_before + timedelta(hours=duration_hours)
+        
+        # Create time-locked proof
+        window_proof = {
+            "window_id": window_id,
+            "not_before": not_before.isoformat() + "Z",
+            "not_after": not_after.isoformat() + "Z",
+            "issuer": "ire_temporal_validator",
+            "signature": self._sign_temporal_window(window_id, not_before, not_after)
+        }
+        
+        return window_proof
+    
+    def validate_within_window(self, 
+                               action_timestamp: str,
+                               window_proof: Dict) -> bool:
+        """Validate action occurred within execution window"""
+        # Verify signature
+        if not self._verify_signature(window_proof):
+            return False
+        
+        # Parse timestamps
+        action_time = datetime.fromisoformat(action_timestamp.replace('Z', '+00:00'))
+        not_before = datetime.fromisoformat(window_proof['not_before'].replace('Z', '+00:00'))
+        not_after = datetime.fromisoformat(window_proof['not_after'].replace('Z', '+00:00'))
+        
+        # Check bounds
+        return not_before <= action_time <= not_after
+    
+    def detect_time_anomalies(self, workflow_executions: List[Dict]) -> List[Dict]:
+        """Detect temporal manipulation patterns"""
+        anomalies = []
+        
+        for i, execution in enumerate(workflow_executions):
+            # Check for reverse time flow
+            if i > 0:
+                prev_time = datetime.fromisoformat(
+                    workflow_executions[i-1]['timestamp'].replace('Z', '+00:00')
+                )
+                curr_time = datetime.fromisoformat(
+                    execution['timestamp'].replace('Z', '+00:00')
+                )
+                if curr_time < prev_time:
+                    anomalies.append({
+                        "type": "reverse_time_flow",
+                        "execution_id": execution['id'],
+                        "anomaly": f"Time went backwards: {curr_time} < {prev_time}"
+                    })
+            
+            # Check execution duration anomalies
+            if 'duration' in execution:
+                expected_duration = self._get_expected_duration(execution['workflow_type'])
+                if execution['duration'] > expected_duration * 2:
+                    anomalies.append({
+                        "type": "suspicious_duration",
+                        "execution_id": execution['id'],
+                        "anomaly": f"Duration {execution['duration']} exceeds expected {expected_duration}"
+                    })
+        
+        return anomalies
+```
+
+Semantic Laundering Defense
+
+```python
+class EpistemicIntegrityGuard:
+    """Defend against semantic laundering attacks"""
+    
+    def __init__(self):
+        self.similarity_clusters = defaultdict(list)
+        self.source_frequency = defaultdict(int)
+        
+    def check_semantic_laundering(self, 
+                                  content_hash: str,
+                                  raw_content: str,
+                                  source_id: str,
+                                  workflow_id: str) -> Dict:
+        """Check for semantic laundering patterns"""
+        
+        # Check source frequency
+        self.source_frequency[source_id] += 1
+        if self.source_frequency[source_id] > 100:  # Threshold
+            return {
+                "risk": "high",
+                "reason": "Excessive submissions from single source",
+                "action": "throttle"
+            }
+        
+        # Check similarity clusters
+        content_vector = self._vectorize(raw_content)
+        similar = self._find_similar(content_vector)
+        
+        if similar:
+            cluster_id = similar[0]['cluster_id']
+            self.similarity_clusters[cluster_id].append({
+                "content_hash": content_hash,
+                "timestamp": datetime.utcnow().isoformat(),
+                "workflow_id": workflow_id
+            })
+            
+            # Check cluster growth rate
+            if len(self.similarity_clusters[cluster_id]) > 10:
+                return {
+                    "risk": "medium",
+                    "reason": "Rapid growth of similar content cluster",
+                    "action": "flag_for_review"
+                }
+        
+        return {"risk": "low", "reason": "No laundering patterns detected"}
+    
+    def _vectorize(self, content: str) -> List[float]:
+        """Create semantic vector (simplified - use real embeddings in production)"""
+        # Simple bag-of-words for demonstration
+        words = content.lower().split()
+        word_counts = Counter(words)
+        vector = [word_counts.get(w, 0) for w in self.vocabulary]
+        return vector
+    
+    def _find_similar(self, vector: List[float], threshold: float = 0.8):
+        """Find similar vectors in existing clusters"""
+        # Simplified similarity search
+        for cluster_id, items in self.similarity_clusters.items():
+            # In production, use proper vector similarity
+            if random.random() > 0.5:  # Placeholder
+                return items
+        return None
+```
+
+IMPLEMENTATION ROADMAP
+
+Phase 1: Core Sovereignty (Weeks 1-2)
+
+1. Implement validator PKI and attestation framework
+2. Deploy threshold abstraction layer
+3. Add time-window enforcement
+4. Basic semantic laundering detection
+
+Phase 2: Epistemic Defense (Weeks 3-4)
+
+1. Full STRIDE-E threat model implementation
+2. Validator quorum calculus integration
+3. Ledger schema hardening
+4. Cross-validation with frozen models
+
+Phase 3: Operational Resilience (Weeks 5-6)
+
+1. Drift detection and alerting
+2. Validator role rotation policies
+3. Recovery procedures for compromise
+4. Full audit trail integration
+
+Phase 4: Community Governance (Weeks 7-8)
+
+1. Validator reputation system
+2. Plurality-based decision frameworks
+3. Cross-interpreter reconciliation
+4. Sovereign identity integration
+
+VERIFICATION PROTOCOL ENHANCEMENT
+
+Daily Sovereignty Check
+
+```python
+def daily_sovereignty_audit():
+    """Daily audit to ensure no boundary violations"""
+    
+    checks = [
+        # 1. Check n8n for epistemic logic
+        scan_n8n_workflows_for_detection_logic(),
+        
+        # 2. Check IRE for orchestration logic  
+        scan_ire_for_scheduling_logic(),
+        
+        # 3. Verify threshold abstraction
+        verify_no_numeric_thresholds_in_n8n(),
+        
+        # 4. Validate time-window adherence
+        verify_all_executions_within_windows(),
+        
+        # 5. Check validator quorum compliance
+        verify_all_commits_have_proper_quorum(),
+        
+        # 6. Detect semantic laundering
+        run_semantic_laundering_detection(),
+        
+        # 7. Verify workflow definition integrity
+        hash_and_verify_workflow_definitions(),
+        
+        # 8. Check for drift
+        compare_with_frozen_model_baseline()
+    ]
+    
+    sovereignty_score = sum(1 for check in checks if check.passed) / len(checks)
+    
+    return {
+        "sovereignty_score": sovereignty_score,
+        "failed_checks": [c for c in checks if not c.passed],
+        "recommendations": generate_remediation_plan(failed_checks)
+    }
+```
+
+Weekly Epistemic Integrity Report
+
+```python
+def weekly_epistemic_integrity_report():
+    """Weekly comprehensive epistemic health report"""
+    
+    report = {
+        "temporal_integrity": {
+            "execution_windows_violated": count_window_violations(),
+            "time_anomalies_detected": detect_time_anomalies(),
+            "average_execution_latency": calculate_latency()
+        },
+        "validator_health": {
+            "active_validators": count_active_validators(),
+            "quorum_satisfaction_rate": calculate_quorum_rate(),
+            "validator_role_diversity": measure_diversity()
+        },
+        "detection_quality": {
+            "drift_from_baseline": measure_drift(),
+            "false_positive_rate": calculate_fpr(),
+            "escalation_accuracy": measure_escalation_accuracy()
+        },
+        "boundary_integrity": {
+            "n8n_epistemic_contamination": detect_contamination(),
+            "ire_orchestration_leakage": detect_leakage(),
+            "workflow_definition_changes": track_workflow_changes()
+        },
+        "semantic_defenses": {
+            "laundering_attempts": count_laundering_attempts(),
+            "similarity_clusters": analyze_clusters(),
+            "source_credibility": assess_source_credibility()
+        }
+    }
+    
+    # Calculate overall epistemic health score
+    report["epistemic_health_score"] = calculate_health_score(report)
+    
+    return report
+```