"""SR01 shared-history component, not a complete consent/chain verifier. No network, persistent storage, live attester or positive CONSENT_CURRENT result. The accompanying C01 dependency is a byte-preserved reference from -03. """ from copy import deepcopy import hashlib import json from pathlib import Path from cryptography.exceptions import InvalidSignature from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey from jsonschema import Draft202012Validator import c01_reference as c PROFILE = 'xcim-consent-sync-01' TYP = 'application/xcim-consent-sync+json' ROLES = {'revocation_log_head': 'revocation_log', 'revocation_checkpoint': 'revocation_checkpoint'} scope_fields = {'profile': {'const': PROFILE}, 'issuer': c.URI, 'sender_id': c.URI, 'sender_epoch': c.POSINT} def shape(kind, fields): return c.closed({**scope_fields, 'kind': {'const': kind}, **fields}) DEFS = { 'genesis': shape('revocation_genesis', {}), 'entry': shape('revocation_entry', {'sequence': c.POSINT, 'previous_entry_commitment': c.HASH, 'event_commitment': c.HASH}), 'head': shape('revocation_log_head', {'sequence': c.UINT, 'event_head': c.HASH, 'issued_at': c.UINT}), 'checkpoint': shape('revocation_checkpoint', {'log_head_commitment': c.HASH, 'covered_at': c.UINT, 'this_update': c.UINT, 'next_update': c.UINT}), 'protected': c.closed({'alg': {'const': 'Ed25519'}, 'kid': c.SCHEMA['$defs']['protected']['properties']['kid'], 'typ': {'const': TYP}}), 'evidence': c.closed({'profile': {'const': PROFILE}, **{k: c.SCHEMA['$defs']['jws'] for k in ('sender', 'consent', 'log_head', 'checkpoint')}, 'consent_anchor': c.SCHEMA['$defs']['anchor'], 'revocation_anchors': c.SCHEMA['$defs']['evidence']['properties']['revocation_anchors']}), } SCHEMA = {'$schema': 'https://json-schema.org/draft/2020-12/schema', '$id': 'https://xcim.org/schemas/sr01/sync.schema.json', '$comment': 'Structural validation only; history, grant, authority and blockchain checks remain mandatory.', '$defs': {**deepcopy(c.SCHEMA['$defs']), **DEFS}, '$ref': '#/$defs/evidence'} def validate(kind, obj): validator = Draft202012Validator({'$defs': SCHEMA['$defs'], '$ref': '#/$defs/' + kind}) c.require(not list(validator.iter_errors(obj)), 'SR_SCHEMA') for key in ('issuer', 'sender_id'): if key in obj: c.require(c.valid_uri(obj[key]), 'SR_IDENTIFIER') for key, value in obj.items(): if key.endswith('_commitment') or key == 'event_head': c.unb64(value, 32) def hs(obj): return c.b64(hashlib.sha256(b'XCIM-SR01-OBJECT\0' + c.jcs(obj)).digest()) def genesis(scope): obj = {'profile': PROFILE, 'kind': 'revocation_genesis', **scope} validate('genesis', obj) return hs(obj) def sign(obj, key, kid): protected = c.b64(c.jcs({'alg': 'Ed25519', 'kid': kid, 'typ': TYP})) payload = c.b64(c.jcs(obj)) message = (protected + '.' + payload).encode('ascii') return {'protected': protected, 'payload': payload, 'signature': c.b64(key.sign(message))} def signed(env, kind, policy): c.validate('jws', env) c.require(len(c.jcs(env)) <= 128 * 1024, 'SR_SIZE') header = c.strict_json(c.unb64(env['protected']), canonical=True) validate('protected', header) obj = c.strict_json(c.unb64(env['payload']), canonical=True) validate(kind, obj) c.require(obj['issuer'] == policy['issuer'], 'SR_ISSUER') key = policy['keys'].get(header['kid']) c.require(key is not None and ROLES[obj['kind']] in key['roles'] and not key.get('revoked', False) and key.get('historical_accepted', True), 'SR_KEY_ROLE') issued = obj.get('issued_at', obj.get('this_update')) c.require(key['not_before'] <= issued < key['not_after'], 'SR_KEY_TIME') try: Ed25519PublicKey.from_public_bytes(c.unb64(key['public_key'], 32)).verify( c.unb64(env['signature'], 64), (env['protected'] + '.' + env['payload']).encode('ascii')) except (InvalidSignature, ValueError) as exc: raise c.C01Error('SR_SIGNATURE') from exc return obj, key class Replica: """Atomic in-memory component demo; disk durability and recovery are unimplemented.""" def __init__(self, scope): self.scope = deepcopy(scope) self.sequence = 0 self.head = genesis(scope) self.entries = [] self.events = {} self.ids = set() self.grant_ids = {} self.known_heads = {} self.known_sequence = 0 self.coverage_floor = 0 self.conflict = False self.last_issued = 0 def same_scope(self, obj): c.require(all(obj[k] == v for k, v in self.scope.items()), 'SR_SCOPE') def learn_head(self, env, policy): obj, _ = signed(env, 'head', policy) self.same_scope(obj) if obj['sequence'] == 0: c.require(obj['event_head'] == genesis(self.scope), 'SR_GENESIS') old = self.known_heads.get(obj['sequence']) if old is not None and old != hs(obj): self.conflict = True raise c.C01Error('SR_HEAD_CONFLICT') self.known_heads[obj['sequence']] = hs(obj) self.known_sequence = max(self.known_sequence, obj['sequence']) return obj def apply_page(self, pairs, policy, now): c.require(isinstance(pairs, list) and len(pairs) <= 1024 and len(c.jcs(pairs)) <= 2 * 1024 * 1024, 'SR_PAGE_LIMIT') stage = deepcopy(self) for pair in pairs: c.require(isinstance(pair, dict) and set(pair) == {'entry', 'revocation'}, 'SR_PAIR') entry = pair['entry'] validate('entry', entry) stage.same_scope(entry) event = c.verify_signed(pair['revocation'], 'revocation', policy) c.require(event['issuer'] == stage.scope['issuer'], 'SR_SCOPE') c.require(entry['event_commitment'] == c.commitment(event), 'SR_EVENT_COMMITMENT') if entry['sequence'] <= stage.sequence: if hs(entry) != stage.entries[entry['sequence'] - 1]: # An unsigned alternative entry alone cannot poison durable state. raise c.C01Error('SR_ENTRY_CONFLICT') continue c.require(entry['sequence'] == stage.sequence + 1 and entry['previous_entry_commitment'] == stage.head, 'SR_GAP') gh = event['consent_commitment'] history = stage.events.get(gh, []) c.require(event['event_id'] not in stage.ids, 'SR_EVENT_DUPLICATE') c.require(stage.grant_ids.get(event['consent_id'], gh) == gh, 'SR_GRANT_ID_CONFLICT') c.require(event['sequence'] == len(history) + 1 and event['previous_event_commitment'] == (c.commitment(history[-1]) if history else gh), 'SR_GRANT_CHAIN') removed = {t for previous in history for t in previous['type_ids']} c.require(c.sorted_unique(event['type_ids']) and not removed.intersection(event['type_ids']), 'SR_TYPE_REINTRODUCTION') c.require(event['method'] in policy['revocation_methods'].get(event['actor'], []), 'SR_METHOD') c.require(event['requested_at'] <= event['effective_at'] <= event['issued_at'] <= now, 'SR_EVENT_TIME') if history: c.require(history[-1]['consent_id'] == event['consent_id'] and history[-1]['effective_at'] <= event['effective_at'] and history[-1]['issued_at'] <= event['issued_at'], 'SR_EVENT_ORDER') stage.events.setdefault(gh, []).append(event) stage.ids.add(event['event_id']) stage.grant_ids[event['consent_id']] = gh stage.sequence += 1 stage.head = hs(entry) stage.entries.append(stage.head) stage.last_issued = max(stage.last_issued, event['issued_at']) self.__dict__.update(stage.__dict__) def check_sync(self, head_env, checkpoint_env, policy, now, grant_issued_at): c.require(policy.get('sr01_admitted') is True, 'SR_NOT_ADMITTED') head = self.learn_head(head_env, policy) cp, key = signed(checkpoint_env, 'checkpoint', policy) self.same_scope(cp) c.require(cp['log_head_commitment'] == hs(head), 'SR_CHECKPOINT_HEAD') c.require(head['issued_at'] <= cp['covered_at'] <= cp['this_update'] < cp['next_update'] and cp['next_update'] - cp['covered_at'] <= 300, 'SR_INTERVAL') c.require(not self.conflict, 'SR_QUARANTINE') c.require(head['sequence'] >= self.known_sequence and head['sequence'] == self.sequence and head['event_head'] == self.head, 'SR_INCOMPLETE_OR_ROLLBACK') c.require(self.last_issued <= head['issued_at'], 'SR_HEAD_TIME') c.require(cp['covered_at'] >= self.coverage_floor, 'SR_COVERAGE_ROLLBACK') u = policy['clock_uncertainty'] c.require(type(u) is int and 0 <= u <= 60, 'SR_CLOCK') c.require(policy['not_before'] <= now-u and now+u < policy['expires_at'], 'SR_POLICY_TIME') c.require(key['not_before'] <= now-u and now+u < key['not_after'], 'SR_KEY_CURRENT') c.require(cp['this_update'] <= now-u and now+u < cp['next_update'] and now+u-cp['covered_at'] <= 300, 'SR_STALE') c.require(grant_issued_at <= cp['covered_at'], 'SR_GRANT_AFTER_CUT') self.coverage_floor = cp['covered_at'] return {'synchronization_checks': 'COMPLETE', 'profile': PROFILE, 'sequence': self.sequence, 'covered_at': cp['covered_at'], 'can_assert_xcim_pass': False, 'consent_result': None, 'required_remaining': ['grant-and-query-verification', 'current-authority', 'grant-relative-withdrawal-validation', 'grant-blockchain-and-withdrawal-anchor-verification', 'original-expiry-and-final-time-check', 'durable-recovery']} if __name__ == '__main__': folder = Path(__file__).resolve().parent / 'schemas' folder.mkdir(exist_ok=True) (folder / 'sync.schema.json').write_text(json.dumps(SCHEMA, indent=2)+'\n', encoding='utf-8') print('Generated SR01 structural schema; no operational consent result.')