Trust recovery and access rollback

"""Offline worksheet: reconcile synthetic binding records, never call IAM.

localId is an exercise identifier, NOT an IAM binding field. Conditions are
opaque text: this program does not parse CEL or calculate effective access.
A proposal needs human review and a fresh real read-modify-write cycle.
"""
from copy import deepcopy
from hashlib import sha256
from itertools import permutations, product
from pathlib import Path
import json


def nonblank(value):
 return isinstance(value, str) and bool(value.strip) and value == value.strip


def validate(snapshot):
 if not isinstance(snapshot, dict) or set(snapshot)!= {'resource', 'etag', 'version', 'bindings'}:
 raise ValueError('Expected exactly resource, etag, version and bindings')
 if not all(nonblank(snapshot[k]) for k in ['resource', 'etag']):
 raise ValueError('Resource and declared etag must be nonblank strings')
 if type(snapshot['version']) is not int or snapshot['version'] not in (1, 3):
 raise ValueError('Worksheet accepts schema version 1 or 3 only')
 if not isinstance(snapshot['bindings'], list):
 raise ValueError('Bindings must be a list')
 indexed = {}
 for binding in snapshot['bindings']:
 if not isinstance(binding, dict) or set(binding)!= {'localId', 'role', 'members', 'condition'}:
 raise ValueError('Expected exact synthetic binding fields')
 if not all(nonblank(binding[k]) for k in ['localId', 'role']):
 raise ValueError('Local ID and role must be nonblank strings')
 if binding['localId'] in indexed:
 raise ValueError('Duplicate local ID')
 members = binding['members']
 if not isinstance(members, list) or not members or not all(nonblank(m) for m in members):
 raise ValueError('Members must be a nonempty list of nonblank strings')
 if len(members)!= len(set(members)):
 raise ValueError('Duplicate member')
 condition = binding['condition']
 if condition is not None and (not nonblank(condition) or snapshot['version']!= 3):
 raise ValueError('Opaque condition text requires schema version 3')
 indexed[binding['localId']] = {**deepcopy(binding), 'members': sorted(members)}
 return indexed


def reconcile(before, after, current):
 """Produce an all-or-nothing structural proposal from declared snapshots.

 IDs touched between before/after define the scope. A touched current record
 must match after (invert it) or before (already reverted); otherwise conflict.
 Unrelated current records survive. Etag is carried as evidence, not verified.
 """
 old, changed, now = [validate(s) for s in (before, after, current)]
 if len({s['resource'] for s in (before, after, current)})!= 1:
 raise ValueError('Snapshots must describe the same declared resource')
 touched = sorted(k for k in old.keys | changed.keys if old.get(k)!= changed.get(k))
 proposal = deepcopy(now)
 decisions = []
 for key in touched:
 if now.get(key) == old.get(key):
 action = 'already-reverted'
 elif now.get(key) == changed.get(key):
 action = 'restore-before'
 if key in old:
 proposal[key] = deepcopy(old[key])
 else:
 proposal.pop(key, None)
 else:
 action = 'conflict'
 decisions.append({'localId': key, 'action': action})
 conflict = any(d['action'] == 'conflict' for d in decisions)
 result = {
 'resource': current['resource'], 'observedEtag': current['etag'],
 'decisions': decisions, 'conflict': conflict,
 'untouchedCurrentIds': sorted(set(now) - set(touched)),
 'proposal': None, 'humanReviewRequired': True,
 'etagFreshnessVerified': False, 'celEvaluated': False,
 'effectiveAccessVerified': False, 'productionWriteAuthorized': False,
 }
 if not conflict:
 bindings = [proposal[k] for k in sorted(proposal)]
 result['proposal'] = {'resource': current['resource'], 'etag': current['etag'],
 'version': 3 if any(b['condition'] is not None for b in bindings) else current['version'],
 'bindings': bindings}
 return result


def binding(key='batch', member='serviceAccount:batch@example.invalid', condition='window-A'):
 return {'localId': key, 'role': 'roles/example.reader', 'members': [member], 'condition': condition}


def snapshot(bindings=None, etag='current-C', version=3):
 return {'resource': 'projects/fictional-funds', 'etag': etag, 'version': version,
 'bindings': deepcopy([binding] if bindings is None else bindings)}


def evidence:
 before = snapshot(etag='before-A')
 after = snapshot([binding(condition=None)], 'after-B')
 concurrent = binding('recovery', 'group:recovery@example.invalid', None)
 fixtures = []
 def record(label, b, a, c):
 original = deepcopy((b, a, c))
 result = reconcile(b, a, c)
 assert (b, a, c) == original
 fixtures.append({'id': label, **result})
 return result
 r = record('preserve-concurrent-grant', before, after, snapshot([binding(condition=None), concurrent]))
 assert r['proposal']['bindings'] == [binding, concurrent]
 assert r['proposal']['etag'] == 'current-C'
 assert r['untouchedCurrentIds'] == ['recovery']
 r = record('same-binding-concurrent-edit', before, after, snapshot([binding(condition='window-C')]))
 assert r['conflict'] and r['proposal'] is None
 r = record('already-reverted', before, after, before)
 assert r['decisions'][0]['action'] == 'already-reverted'
 r = record('restore-deleted-binding', before, snapshot([]), snapshot([concurrent]))
 assert r['proposal']['bindings'] == [binding, concurrent]
 r = record('remove-added-binding', snapshot([]), before, snapshot([binding, concurrent]))
 assert r['proposal']['bindings'] == [concurrent]
 r = record('deleted-by-another-change', before, after, snapshot([]))
 assert r['conflict']
 r = record('version-three-restored', before, after, snapshot([binding(condition=None)], version=1))
 assert r['proposal']['version'] == 3
 r = record('no-declared-binding-change', before, before, snapshot([concurrent]))
 assert r['decisions'] == [] and r['proposal']['bindings'] == [concurrent]
 multi_before = snapshot([binding, binding('second')])
 multi_after = snapshot([binding(condition=None), binding('second', condition=None)])
 r = record('atomic-conflict', multi_before, multi_after,
 snapshot([binding(condition=None), binding('second', condition='other-change')]))
 assert r['proposal'] is None
 r = record('opaque-condition-change', before, after, snapshot([binding(condition='window-A && true')]))
 assert r['conflict'] # No claim that logically equivalent expressions compare equal.
 r = record('member-order-only', snapshot([{'localId':'x','role':'roles/example.reader','members':['user:a','user:b'],'condition':None}]),
 snapshot([{'localId':'x','role':'roles/example.reader','members':['user:b','user:a'],'condition':None}]), snapshot([]))
 assert r['decisions'] == []
 r = record('no-condition-version-preserved', snapshot([], version=1), snapshot([concurrent]), snapshot([concurrent]))
 assert r['proposal']['version'] == 3
 states = [None, binding, binding(condition=None), binding(condition='window-C')]
 combinations = 0
 for b, a, c in product(states, repeat=3):
 inputs = [snapshot([] if x is None else [x]) for x in (b, a, c)]
 result = reconcile(*inputs)
 expected_conflict = b!= a and c!= b and c!= a
 assert result['conflict'] == expected_conflict
 if not expected_conflict:
 expected = c if b == a else b
 assert result['proposal']['bindings'] == ([] if expected is None else [expected])
 combinations += 1
 rows = [binding(condition=None), concurrent, binding('observer', condition=None)]
 expected = reconcile(before, after, snapshot(rows))
 orders = 0
 for order in permutations(rows):
 assert reconcile(before, after, snapshot(list(order))) == expected
 orders += 1
 invalid = []
 def altered(fn):
 s = snapshot; fn(s); invalid.append(s)
 altered(lambda s:s.pop('etag'))
 altered(lambda s:s.update(etag=''))
 altered(lambda s:s.update(etag=' stale '))
 altered(lambda s:s.update(resource=''))
 altered(lambda s:s.update(version=True))
 altered(lambda s:s.update(version=2))
 altered(lambda s:s.update(version=1))
 altered(lambda s:s.update(bindings={}))
 altered(lambda s:s['bindings'].append(deepcopy(s['bindings'][0])))
 altered(lambda s:s['bindings'][0].update(localId=''))
 altered(lambda s:s['bindings'][0].update(role=''))
 altered(lambda s:s['bindings'][0].update(members=[]))
 altered(lambda s:s['bindings'][0].update(members=['user:a','user:a']))
 altered(lambda s:s['bindings'][0].update(members=['']))
 altered(lambda s:s['bindings'][0].update(condition=''))
 altered(lambda s:s['bindings'][0].update(condition={'expression':'true'}))
 altered(lambda s:s['bindings'][0].update(extra='unsupported'))
 altered(lambda s:s.update(auditConfigs=[]))
 invalid.extend([None, [], 'policy'])
 for value in invalid:
 try:
 reconcile(before, after, value)
 except ValueError:
 pass
 else:
 raise AssertionError('Invalid input accepted')
 other = snapshot; other['resource'] = 'projects/other'
 try:
 reconcile(before, after, other)
 except ValueError:
 pass
 else:
 raise AssertionError('Resource mismatch accepted')
 return {'scriptSha256': sha256(Path(__file__).read_bytes).hexdigest,
 'fixtures': fixtures, 'stateCombinations': combinations, 'inputPermutations': orders,
 'invalidInputs': len(invalid)+1, 'inputPreserved': True, 'orderIndependent': True,
 'network': False, 'cloudExecuted': False, 'persistentWrites': False}


if __name__ == '__main__':
 print(json.dumps(evidence, ensure_ascii=False, indent=2))

Related topics: Revocation, sessions and recovery decisions · Identities, credentials and access evidence · Governance, scope and control evidence

Reference: Recover an account protected by 2-Step Verification · Current linked guide; edition date unconfirmed (2026-09-30 inspection)