Système immunitaire du graphe

import json, os, pathlib, time
import requests
import urllib3

# The quick-start server uses a self-signed certificate; disable verification
# for the local playbook only, never against a real deployment.
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

BASE_URL = os.environ.get('CORROBORE_URL', 'https://127.0.0.1:8080')
TOKEN = os.environ.get('CORROBORE_HTTP_AUTH_TOKEN', 'change-me')

http = requests.Session()
http.verify = False
http.headers.update({'Authorization': f'Bearer {TOKEN}', 'Content-Type': 'application/json'})


class MemoryError_(RuntimeError):
    """Carries the stable v1 error taxonomy instead of a bare HTTP status."""

    def __init__(self, code, message, status):
        super().__init__(f'{code}: {message}')
        self.code, self.message, self.status = code, message, status


def memory_op(operation, payload, idempotency_key=None, expect_error=False):
    """POST /v1/memory/operations and unwrap the typed result."""
    body = {'contract_version': 'v1', 'operation': operation, 'input': payload}
    if idempotency_key is not None:
        body['idempotency_key'] = idempotency_key
    for attempt in range(12):
        response = http.post(f'{BASE_URL}/v1/memory/operations', data=json.dumps(body))
        # Protected routes share a global token bucket (50 rps sustained, 200 burst by
        # default) that a bulk load will hit. Two details matter here: the 429 body is
        # plain text, not the JSON error envelope, and `Retry-After` can be `0` — so
        # honour it as a floor, never as the whole wait.
        if response.status_code == 429:
            hinted = float(response.headers.get('Retry-After', 0) or 0)
            time.sleep(max(hinted, 0.2 * (attempt + 1)))
            continue
        break
    if response.status_code != 200:
        try:
            error = response.json().get('error', {})
        except ValueError:  # 429 and other transport rejections are not JSON
            error = {}
        failure = MemoryError_(error.get('code', 'UNKNOWN'), error.get('message', response.text), response.status_code)
        if expect_error:
            return failure
        raise failure
    if expect_error:
        raise AssertionError(f'{operation} unexpectedly succeeded')
    return response.json()['result']['result']


ready = http.get(f'{BASE_URL}/health/ready').json()
version = http.get(f'{BASE_URL}/version').json()
print('ready  :', json.dumps(ready)[:160])
print('version:', json.dumps(version)[:160])

from collections import Counter

survey = memory_op('recall', {
    'objective': 'integrity indicator',
    'seed_ids': [ids['case--containment']],
    'limits': LIMITS,
})
records = {i['record']['id']: i['record'] for i in survey['items']}

degree = Counter()
for edge in survey['relationships']:
    degree[edge['source_id']] += 1
    degree[edge['target_id']] += 1

ranked = [(by_id.get(node, node), count) for node, count in degree.most_common(5)]
print(f"{'record':<34}{'degree':>8}")
print('-' * 42)
for key, count in ranked:
    print(f'{key:<34}{count:>8}')

outliers = {key for key, count in ranked[:2]}
print()
print('structural outliers:', sorted(outliers))
assert 'indicator--poisoned' in outliers
assert 'hub--legitimate-registrar' in outliers, 'a degree rule alone would flag the registrar too'

contradicted = {}
for edge in survey['relationships']:
    if edge['kind'] == 'contradicts':
        target = records[edge['target_id']]['identity_key']
        origin = records[edge['source_id']]
        contradicted[target] = {
            'by': origin['identity_key'],
            'reliability': origin['content']['value']['properties'].get('source_reliability'),
            'confidence': edge['confidence'],
        }

for target, detail in contradicted.items():
    print(f"{target} contradicted by {detail['by']}")
    print(f"    source reliability {detail['reliability']}, edge confidence {detail['confidence']}")

assert 'indicator--poisoned' in contradicted
assert 'hub--legitimate-registrar' not in contradicted

def immune_findings(records, degree, contradicted, feeds, degree_threshold=20, rate_threshold=0.2):
    findings = []
    for record in records.values():
        key = record['identity_key']
        if record['kind'] != 'integrity.indicator':
            continue
        signals = []
        if degree.get(ids[key], 0) >= degree_threshold:
            signals.append(f'structural: degree {degree[ids[key]]}')
        if key in contradicted:
            signals.append(f"epistemic: contradicted by {contradicted[key]['by']}")
        asserted_by = record['content']['value']['properties'].get('asserted_by')
        rate = feeds.get(asserted_by, {}).get('retraction_rate')
        if rate is not None and rate >= rate_threshold:
            signals.append(f'behavioural: {asserted_by} retraction rate {rate}')
        if len(signals) >= 3:
            findings.append({'record': key, 'signals': signals, 'action': 'quarantine'})
        elif len(signals) == 2:
            findings.append({'record': key, 'signals': signals, 'action': 'request_review'})
    return findings


findings = immune_findings(records, degree, contradicted, feeds)
for finding in findings:
    print(f"{finding['record']}  ->  {finding['action']}")
    for signal in finding['signals']:
        print(f'    {signal}')

quarantine = [f['record'] for f in findings if f['action'] == 'quarantine']
assert quarantine == ['indicator--poisoned']

target = quarantine[0]
receipt = memory_op('forget', {
    'memory_id': ids[target],
    'mode': 'tombstone',
    'reason': 'immune finding: structural, epistemic and behavioural signals correlated',
    'expires_at': None,
}, idempotency_key='imm:quarantine')
print('mode   :', receipt['mode'])
print('receipt:', receipt['receipt'])

after = memory_op('recall', {
    'objective': 'integrity indicator',
    'seed_ids': [ids['case--containment']],
    'limits': LIMITS,
})
still_returned = {i['record']['identity_key'] for i in after['items']}

print()
print('poisoned pivot still retrieved:', target in still_returned)
print('legitimate hub still retrieved:', 'hub--legitimate-registrar' in still_returned)
print('indicators still retrieved    :',
      sum(1 for k in still_returned if k.startswith('indicator--')))

assert target not in still_returned
assert 'hub--legitimate-registrar' in still_returned