Fehlende Quellenkorrelation mehrerer unabhängiger Daten

Beschreibung

Fehlende Quellenkorrelation mehrerer unabhängiger Daten tritt auf, wenn ein Produkt sich auf eine einzelne Datenquelle verlässt und dadurch die Möglichkeit verhindert, zu erkennen, ob ein Angreifer diese Datenquelle kompromittiert hat. Während implizite Signierung sicherstellen kann, dass Daten während der Übertragung nicht manipuliert wurden, kann sie nicht garantieren, dass die Quelle selbst nicht kompromittiert wurde. Die Gegenmaßnahme besteht darin, mehrere unabhängige Quellen abzufragen und Antworten zu vergleichen. Wenn Antworten voneinander abweichen, sollte das System potenziell kompromittierte Quellen kennzeichnen. Wenn nicht genügend Antworten vorhanden sind, um eine Mehrheit oder Pluralität festzustellen, sollten alle Quellen als potenziell kompromittiert gemeldet werden. Die Anzahl unabhängiger Quellen sollte mit der Schwere der Auswirkungen von Integritätsfehlern skalieren.

Risiko

Abhängigkeit von einer einzelnen Quelle hat schwerwiegende Sicherheitsauswirkungen. Single-Point-of-Failure-Angriffe ermöglicht. Person-in-the-Middle-Angriffe können Validierung unterwandern. Beliebige Antworten ohne Erkennung einschleusbar. Datenintegrität nicht verifizierbar. Kompromittierte Quellen bleiben unentdeckt. Keine Resilienz gegen Oracle-Manipulation. Kritische Entscheidungen basierend auf potenziell gefälschten Daten. Hohe Auswirkung für sicherheitskritische Anwendungen wie Zertifikatsvalidierung.

Lösung

Implementieren Sie Practical-Byzantine-Fault-Tolerance-Methoden, um Daten von mehreren Quellen anzufordern, Konsistenz zu überprüfen und potenziell kompromittierte Quellen zu melden. Vermeiden Sie Abhängigkeit von einer einzelnen Quelle. Etablieren Sie Meldemechanismen für kompromittierte Quellen. Stellen Sie sicher, dass Informationsquellen wirklich unabhängig sind (keine gemeinsame Infrastruktur). Integrieren Sie die Meldung abweichender Antworten in Incident-Response-Verfahren. Skalieren Sie die Anzahl der Quellen mit der Kritikalität der Daten.

Häufige Auswirkungen

Auswirkung	Details
Vertraulichkeit	Bereich: Vertraulichkeit Angreifer kann Anwendungsdaten lesen, indem er vertrauenswürdige Quellen imitiert.
Integrität	Bereich: Integrität Angreifer kann Anwendungsdaten durch kompromittierte Quelle modifizieren.
Zugriffskontrolle	Bereich: Zugriffskontrolle Privilegieneskalation oder Identitätsübernahme über gefälschte Antworten.

Beispielcode und Lösung

Verwundbarer Code

# VERWUNDBAR: Abhängigkeit von einzelner Quelle für kritische Daten

import requests
from cryptography import x509
from datetime import datetime

class VulnerableCertificateValidator:
    def __init__(self):
        # VERWUNDBAR: Einzelner OCSP-Responder
        self.ocsp_endpoint = "https://ocsp.example.com"

    def check_certificate_revocation(self, cert):
        """Prüfen ob Zertifikat widerrufen - VERWUNDBAR: Einzelne Quelle."""
        # VERWUNDBAR: Nur eine Quelle abgefragt
        response = requests.post(
            self.ocsp_endpoint,
            data=self._build_ocsp_request(cert),
            headers={'Content-Type': 'application/ocsp-request'}
        )

        # VERWUNDBAR: Antwort ohne Verifizierung vertrauen
        if response.status_code == 200:
            return self._parse_ocsp_response(response.content)

        # VERWUNDBAR: Bei Fehler offen versagen
        return True  # Annehmen, nicht widerrufen wenn Prüfung fehlschlägt

        # Angriffsszenario:
        # 1. Angreifer kompromittiert oder imitiert OCSP-Responder
        # 2. Angreifer gibt "nicht widerrufen" für widerrufene Zertifikate zurück
        # 3. Anwendung vertraut kompromittierten Zertifikaten
        # 4. Keine Möglichkeit, den Angriff mit einzelner Quelle zu erkennen

class VulnerableDNSResolver:
    def __init__(self):
        # VERWUNDBAR: Einzelner DNS-Server
        self.dns_server = "8.8.8.8"

    def resolve(self, domain):
        """Domain auflösen - VERWUNDBAR: Einzelne Quelle."""
        import dns.resolver

        # VERWUNDBAR: Nur einen Server abfragen
        resolver = dns.resolver.Resolver()
        resolver.nameservers = [self.dns_server]

        try:
            answer = resolver.resolve(domain, 'A')
            return [str(rdata) for rdata in answer]
        except Exception:
            return []

        # Angriff: DNS-Poisoning auf einzelnem Server betrifft alle Abfragen

class VulnerableTimeSync:
    def __init__(self):
        # VERWUNDBAR: Einzelner NTP-Server
        self.ntp_server = "time.example.com"

    def get_current_time(self):
        """Aktuelle Zeit abrufen - VERWUNDBAR: Einzelne Quelle."""
        import ntplib

        client = ntplib.NTPClient()

        # VERWUNDBAR: Einzelner Zeitquelle vertrauen
        response = client.request(self.ntp_server)
        return datetime.fromtimestamp(response.tx_time)

        # Angriff: Kompromittierter NTP-Server kann Zeit manipulieren
        # Dies betrifft Zertifikatsvalidierung, Log-Zeitstempel usw.

// VERWUNDBAR: Einzelquellen-Blockchain-Oracle

class VulnerableOracle {
    constructor() {
        // VERWUNDBAR: Einzelnes Preis-Oracle
        this.oracleEndpoint = "https://oracle.example.com/price";
    }

    async getAssetPrice(asset) {
        // VERWUNDBAR: Einzelne Quelle für kritische Finanzdaten
        const response = await fetch(`${this.oracleEndpoint}/${asset}`);
        const data = await response.json();

        // VERWUNDBAR: Ohne Verifizierung vertrauen
        return data.price;

        // Angriff: Oracle-Manipulation kann verursachen:
        // - Fehlerhafte Liquidationen in DeFi
        // - Preismanipulation für Arbitrage
        // - Diebstahl von Mitteln basierend auf falschen Preisen
    }

    async executeTrade(asset, threshold) {
        const price = await this.getAssetPrice(asset);

        // VERWUNDBAR: Entscheidung basierend auf einzelner Quelle
        if (price < threshold) {
            // Trade basierend auf potenziell manipuliertem Preis ausführen
            return this.buy(asset, price);
        }
    }
}

// VERWUNDBAR: Einzelne API-Schlüssel-Validierungsstelle
class VulnerableAuthValidator {
    constructor() {
        this.authEndpoint = "https://auth.example.com/validate";
    }

    async validateToken(token) {
        // VERWUNDBAR: Einzelne Validierungsquelle
        const response = await fetch(this.authEndpoint, {
            method: 'POST',
            body: JSON.stringify({ token }),
            headers: { 'Content-Type': 'application/json' }
        });

        // VERWUNDBAR: Antwort vollständig vertrauen
        return response.ok;

        // Angriff: Kompromittierter Auth-Server erlaubt ungültige Token
    }
}

Sichere Lösung

# SICHER: Mehrquellen-Verifizierung mit Byzantinischer Fehlertoleranz

import requests
import hashlib
from typing import List, Dict, Any, Optional, Tuple
from collections import Counter
from dataclasses import dataclass
from datetime import datetime
import asyncio
import aiohttp

@dataclass
class SourceResponse:
    source: str
    data: Any
    timestamp: datetime
    signature: Optional[bytes] = None
    error: Optional[str] = None

class SecureCertificateValidator:
    def __init__(self):
        # SICHER: Mehrere unabhängige OCSP-Responder
        self.ocsp_endpoints = [
            "https://ocsp1.example.com",
            "https://ocsp2.example.com",
            "https://ocsp3.example.com",
            "https://ocsp4.different-provider.com",  # Anderer Anbieter!
            "https://ocsp5.another-provider.com"     # Weiterer Anbieter!
        ]
        # SICHER: Minimaler erforderlicher Konsens
        self.min_consensus = 3
        self.max_divergence = 1  # Max. abweichende Quellen vor Alarm

    async def check_certificate_revocation(self, cert) -> Tuple[bool, List[str]]:
        """Zertifikatswiderruf mit Mehrquellen-Verifizierung prüfen."""
        # SICHER: Alle Quellen parallel abfragen
        responses = await self._query_all_sources(cert)

        # SICHER: Antworten auf Konsens analysieren
        valid_responses = [r for r in responses if r.error is None]

        if len(valid_responses) < self.min_consensus:
            # SICHER: Nicht genug Antworten - sicher versagen
            await self._alert_insufficient_sources(responses)
            return False, ["Unzureichende Quellenantworten"]

        # SICHER: Auf Konsens prüfen
        revocation_status = Counter(r.data['revoked'] for r in valid_responses)

        # Mehrheitsmeinung ermitteln
        majority_status, majority_count = revocation_status.most_common(1)[0]
        minority_count = len(valid_responses) - majority_count

        compromised_sources = []

        # SICHER: Potenziell kompromittierte Quellen erkennen
        if minority_count > 0:
            minority_status = not majority_status
            compromised_sources = [
                r.source for r in valid_responses
                if r.data['revoked'] == minority_status
            ]

            if minority_count > self.max_divergence:
                # SICHER: Zu viel Abweichung - alle als verdächtig melden
                await self._alert_divergence(responses, majority_count, minority_count)

        return majority_status, compromised_sources

    async def _query_all_sources(self, cert) -> List[SourceResponse]:
        """Alle OCSP-Quellen parallel abfragen."""
        async with aiohttp.ClientSession() as session:
            tasks = [
                self._query_source(session, endpoint, cert)
                for endpoint in self.ocsp_endpoints
            ]
            return await asyncio.gather(*tasks)

    async def _query_source(self, session, endpoint, cert) -> SourceResponse:
        """Eine einzelne OCSP-Quelle abfragen."""
        try:
            async with session.post(
                endpoint,
                data=self._build_ocsp_request(cert),
                timeout=aiohttp.ClientTimeout(total=5)
            ) as response:
                if response.status == 200:
                    content = await response.read()
                    return SourceResponse(
                        source=endpoint,
                        data=self._parse_ocsp_response(content),
                        timestamp=datetime.utcnow()
                    )
                else:
                    return SourceResponse(
                        source=endpoint,
                        data=None,
                        timestamp=datetime.utcnow(),
                        error=f"HTTP {response.status}"
                    )
        except Exception as e:
            return SourceResponse(
                source=endpoint,
                data=None,
                timestamp=datetime.utcnow(),
                error=str(e)
            )

    async def _alert_divergence(self, responses, majority, minority):
        """Sicherheitsteam über Quellenabweichung alarmieren."""
        alert_data = {
            "type": "OCSP_SOURCE_DIVERGENCE",
            "majority_count": majority,
            "minority_count": minority,
            "responses": [
                {"source": r.source, "data": r.data, "error": r.error}
                for r in responses
            ]
        }
        await send_security_alert(alert_data)

class SecureDNSResolver:
    def __init__(self):
        # SICHER: Mehrere unabhängige DNS-Resolver
        self.dns_servers = [
            "8.8.8.8",        # Google
            "1.1.1.1",        # Cloudflare
            "9.9.9.9",        # Quad9
            "208.67.222.222", # OpenDNS
        ]
        self.consensus_threshold = 3

    def resolve(self, domain: str) -> Tuple[List[str], bool]:
        """Domain mit Mehrquellen-Verifizierung auflösen."""
        import dns.resolver

        all_results = {}

        # SICHER: Alle DNS-Server abfragen
        for server in self.dns_servers:
            resolver = dns.resolver.Resolver()
            resolver.nameservers = [server]

            try:
                answer = resolver.resolve(domain, 'A')
                ips = frozenset(str(rdata) for rdata in answer)
                all_results[server] = ips
            except Exception as e:
                all_results[server] = None

        # SICHER: Konsens finden
        valid_results = {k: v for k, v in all_results.items() if v is not None}

        if len(valid_results) < self.consensus_threshold:
            return [], False

        # SICHER: Ergebnisse übereinstimmen
        ip_sets = list(valid_results.values())
        consensus_ips = ip_sets[0]

        all_agree = all(ips == consensus_ips for ips in ip_sets)

        if not all_agree:
            # SICHER: Abweichung erkannt - melden und Schnittmenge finden
            self._alert_dns_divergence(domain, valid_results)
            consensus_ips = set.intersection(*[set(ips) for ips in ip_sets])

        return list(consensus_ips), all_agree

// SICHER: Mehrquellen-Oracle mit Byzantinischer Fehlertoleranz

class SecureOracle {
    constructor() {
        // SICHER: Mehrere unabhängige Oracles
        this.oracles = [
            { url: "https://oracle1.chainlink.com/price", weight: 1 },
            { url: "https://oracle2.band-protocol.com/price", weight: 1 },
            { url: "https://oracle3.api3.org/price", weight: 1 },
            { url: "https://oracle4.tellor.io/price", weight: 1 },
            { url: "https://oracle5.uma.xyz/price", weight: 1 }
        ];

        // SICHER: Byzantinische Fehlertoleranz-Parameter
        this.minResponses = 3;
        this.maxDeviation = 0.02; // 2% maximale Preisabweichung
    }

    async getAssetPrice(asset) {
        // SICHER: Alle Oracles parallel abfragen
        const responses = await Promise.allSettled(
            this.oracles.map(oracle => this.queryOracle(oracle, asset))
        );

        // SICHER: Erfolgreiche Antworten filtern
        const validPrices = responses
            .filter(r => r.status === 'fulfilled' && r.value !== null)
            .map(r => r.value);

        if (validPrices.length < this.minResponses) {
            throw new Error(`Unzureichende Oracle-Antworten: ${validPrices.length}`);
        }

        // SICHER: Konsens über Median prüfen
        const sortedPrices = validPrices.map(p => p.price).sort((a, b) => a - b);
        const medianPrice = sortedPrices[Math.floor(sortedPrices.length / 2)];

        // SICHER: Ausreißer erkennen
        const outliers = validPrices.filter(p => {
            const deviation = Math.abs(p.price - medianPrice) / medianPrice;
            return deviation > this.maxDeviation;
        });

        if (outliers.length > 0) {
            // SICHER: Bei potenzieller Manipulation alarmieren
            await this.alertOutliers(asset, outliers, medianPrice);
        }

        // SICHER: Median-Preis verwenden (resistent gegen Ausreißer)
        return {
            price: medianPrice,
            sources: validPrices.length,
            outliers: outliers.length,
            confidence: this.calculateConfidence(validPrices, medianPrice)
        };
    }

    async queryOracle(oracle, asset) {
        try {
            const response = await fetch(`${oracle.url}/${asset}`, {
                timeout: 5000
            });

            if (!response.ok) return null;

            const data = await response.json();
            return {
                source: oracle.url,
                price: data.price,
                timestamp: data.timestamp,
                weight: oracle.weight
            };
        } catch (e) {
            return null;
        }
    }

    calculateConfidence(prices, median) {
        // SICHER: Konfidenz basierend auf Preisübereinstimmung berechnen
        const deviations = prices.map(p =>
            Math.abs(p.price - median) / median
        );
        const avgDeviation = deviations.reduce((a, b) => a + b) / deviations.length;

        // Höhere Übereinstimmung = höhere Konfidenz
        return Math.max(0, 1 - (avgDeviation * 10));
    }

    async executeTrade(asset, threshold) {
        const priceData = await this.getAssetPrice(asset);

        // SICHER: Konfidenz vor Ausführung prüfen
        if (priceData.confidence < 0.8) {
            throw new Error(`Unzureichende Preiskonfidenz: ${priceData.confidence}`);
        }

        // SICHER: Ausreißer-Anzahl prüfen
        if (priceData.outliers > 1) {
            throw new Error(`Zu viele Preis-Ausreißer: ${priceData.outliers}`);
        }

        if (priceData.price < threshold) {
            return this.buy(asset, priceData.price);
        }
    }
}

---

CVE-Beispiele

• CVE-2020-8286: cURLs OCSP-Stapling-Funktion verließ sich auf eine einzelne Quelle ohne Multi-Perspektiven-Validierung.
• CVE-2018-16875: DNS-Resolver vertraute einem einzelnen Upstream-Server, was Cache-Poisoning ermöglichte.

---

Verwandte CWEs

• CWE-345: Unzureichende Verifizierung der Datenauthentizität (übergeordnet)
• CWE-654: Abhängigkeit von einem einzelnen Faktor bei einer Sicherheitsentscheidung (gleichrangig)
• CWE-300: Kanal zugänglich durch Nicht-Endpunkt (verwandt)
• CWE-295: Unsachgemäße Zertifikatsvalidierung (verwandt)

---

Referenzen

1. MITRE Corporation. "CWE-1293: Missing Source Correlation of Multiple Independent Data." https://cwe.mitre.org/data/definitions/1293.html
2. Let's Encrypt. "Multi-Perspective Validation"
3. Castro, M., & Liskov, B. "Practical Byzantine Fault Tolerance" (2002)