#!/usr/bin/env python3 """S04 executable teaching models. Python 3 standard library only; no networking. Run: python foundation-routing-state-check.py > routing-results.json The exhaustive checks use assert; optimized execution is explicitly rejected. """ import itertools import json import math import random from collections import deque from ipaddress import IPv4Address, IPv4Network def bridge_step(fdb, t, ingress, source, destination, ports=(1, 2, 3), ttl=10): """A valid unicast source; '*' denotes broadcast. Mutates only this FDB.""" assert ingress in ports and ttl > 0 for mac, (_, last) in list(fdb.items()): if t - last >= ttl: del fdb[mac] fdb[source] = (ingress, t) if destination != '*' and destination in fdb: out = fdb[destination][0] return [] if out == ingress else [out] return [p for p in ports if p != ingress] def arp_receive(cache, interface, own_ip, own_mac, sender_ip, sender_mac, target_ip, opcode): """RFC826 core merge/target ordering after type/address-format checks.""" key = (interface, sender_ip) known = key in cache if known: cache[key] = sender_mac if target_ip != own_ip: return None if not known: cache[key] = sender_mac if opcode == 1: return dict(opcode=2, source_ip=own_ip, source_mac=own_mac, target_ip=sender_ip, target_mac=sender_mac, ethernet_destination=sender_mac) return None def route(destination, table): candidates = [(IPv4Network(prefix).prefixlen, interface, next_hop) for prefix, interface, next_hop in table if IPv4Address(destination) in IPv4Network(prefix)] if not candidates: return None _, interface, next_hop = max(candidates) return interface, destination if next_hop is None else next_hop def floyd(n, edges): d = [[0 if u == v else math.inf for v in range(n)] for u in range(n)] for (u, v), c in edges.items(): d[u][v] = min(d[u][v], c) for k in range(n): for u in range(n): for v in range(n): d[u][v] = min(d[u][v], d[u][k] + d[k][v]) return d def bridge_tree(n, edges): """Fresh synchronous root/cost advertisements, one simple point-to-point link per pair.""" state = [(u, 0, None) for u in range(n)] rounds = 0 while True: nxt = [] for u in range(n): choices = [(u, 0, -1, -1, -1, None)] # own-root candidate for (v, w), cost in edges.items(): if w == u: # local and remote port IDs use the neighboring bridge ID. choices.append((state[v][0], state[v][1] + cost, v, u, v, v)) best = min(choices) nxt.append((best[0], best[1], best[5])) rounds += 1 if nxt == state: break state = nxt assert rounds <= n + 1 active = set() designated = {} for (u, v), cost in edges.items(): if u < v: winner = min((state[u][0], state[u][1], u, v), (state[v][0], state[v][1], v, u))[2] designated[(u, v)] = winner u_forward = winner == u or state[u][2] == v v_forward = winner == v or state[v][2] == u if u_forward and v_forward: active.add((u, v)) return state, active, designated, rounds def sync_dv(n, edges, destination, infinity=math.inf): d = [infinity] * n d[destination] = 0 history = [d[:]] for _ in range(n - 1): d = [0 if u == destination else min( [infinity] + [min(infinity, c + history[-1][v]) for (a, v), c in edges.items() if a == u]) for u in range(n)] history.append(d[:]) return history def distance_failure(poison=False, three=False): """Atomic local recomputation followed by delivery of its new vector to all peers. Captures one legal sequential execution, not all asynchronous schedules. """ if three: edges = {(0, 1): 1, (1, 0): 1, (1, 2): 1, (2, 1): 1, (0, 2): 1, (2, 0): 10} d, parent = [1, 2, 3], [None, 0, 1] else: edges = {(0, 1): 1, (1, 0): 1} d, parent = [1, 2], [None, 0] n = len(d) def advertisement(v, u): return 16 if poison and parent[v] == u else d[v] cache = {(u, v): advertisement(v, u) for u, v in edges} trace = [] # Direct origin edge of router 0 has now failed; no direct candidate remains. for step in range(80): u = step % n candidates = [(min(16, c + cache[u, v]), v) for (a, v), c in edges.items() if a == u] value, via = min(candidates) d[u], parent[u] = value, via if value < 16 else None for v in range(n): if (v, u) in edges: cache[v, u] = advertisement(u, v) trace.append(dict(router=u, value=value, parent=parent[u], distances=d[:])) if all(x == 16 for x in d): return trace raise AssertionError('failure example failed to reach its finite saturation') def ls_accept(db, origin, sequence, neighbors): old = db.get(origin) if old is not None and sequence <= old[0]: return False db[origin] = (sequence, dict(neighbors)) return True def ls_graph(db): edges = {} for u, (_, neighbors) in db.items(): for v, c in neighbors.items(): if v in db and db[v][1].get(u) == c and c > 0: edges[u, v] = c return edges def path_to(n, edges, source, target): """Enumerate all simple paths as a small-graph oracle for the LS example.""" paths = [] def visit(u, seen, cost): if u == target: paths.append((cost, tuple(seen))) return for (a, v), c in sorted(edges.items()): if a == u and v not in seen: visit(v, seen + [v], cost + c) visit(source, [source], 0) return min(paths) if paths else (math.inf, ()) def policy_choice(node, current, prefer_indirect=True): successor = node % 3 + 1 direct = (node, 0) indirect = (node, successor, 0) candidates = [direct] if current[successor] == (successor, 0): candidates.append(indirect) return indirect if prefer_indirect and indirect in candidates else direct def main(): if not __debug__: raise RuntimeError('This checker requires assertion checks; run Python without -O.') out = {'status': 'PASS', 'models': 'teaching states; no packets are sent'} fdb = {} trace = [] frames = [(0,1,'A','B'), (1,2,'B','A'), (4,1,'A','R'), (5,3,'R','A'), (6,2,'A','B'), (12,3,'R','B')] for args in frames: outputs = bridge_step(fdb, *args) trace.append(dict(time=args[0], outputs=outputs, fdb=dict(fdb))) assert [x['outputs'] for x in trace] == [[2,3],[1],[2,3],[1],[],[1,2]] assert fdb == {'A': (2,6), 'R': (3,12)} # Independent latest-source history oracle over random traces. rng = random.Random(20261009) bridge_checks = 0 for _ in range(500): fdb, history, t = {}, [], 0 for _ in range(40): t += rng.randrange(4) ingress, source, dest = rng.randrange(1,4), rng.choice('ABCR'), rng.choice('ABCR*') history.append((t, ingress, source)) last = {} for time, port, mac in history: if t - time < 10: last[mac] = (port, time) expected = [p for p in (1,2,3) if p != ingress] if dest == '*' or dest not in last else ([] if last[dest][0] == ingress else [last[dest][0]]) assert bridge_step(fdb,t,ingress,source,dest) == expected assert fdb == last bridge_checks += 1 out['learning_bridge'] = {'main_trace': trace, 'independent_history_steps': bridge_checks} a_table = [('192.0.2.0/24','left',None),('0.0.0.0/0','left','192.0.2.1')] r_table = [('192.0.2.0/24','left',None),('198.51.100.0/24','right',None)] assert route('198.51.100.20',a_table) == ('left','192.0.2.1') assert route('198.51.100.20',r_table) == ('right','198.51.100.20') caches = {'A': {}, 'R': {}, 'B': {}} rr = arp_receive(caches['R'],'left','192.0.2.1','MRL','192.0.2.10','MA','192.0.2.1',1) assert rr['ethernet_destination'] == 'MA' arp_receive(caches['A'],'left','192.0.2.10','MA','192.0.2.1','MRL','192.0.2.10',2) br = arp_receive(caches['B'],'right','198.51.100.20','MB','198.51.100.1','MRR','198.51.100.20',1) assert br['ethernet_destination'] == 'MRR' arp_receive(caches['R'],'right','198.51.100.1','MRR','198.51.100.20','MB','198.51.100.1',2) arp_cases = 0 for known, mine, request, other_interface in itertools.product((False,True),repeat=4): key = ('left','peer') cache = ({key:'OLD'} if known else {}) if other_interface: cache[('right','peer')] = 'OTHER' answer = arp_receive(cache,'left','self','SELF','peer','NEW','self' if mine else 'other',1 if request else 2) assert (key in cache) == (known or mine) if key in cache: assert cache[key] == 'NEW' if other_interface: assert cache[('right','peer')] == 'OTHER' assert (answer is not None) == (mine and request) arp_cases += 1 out['arp'] = {'request_targets':['192.0.2.1','198.51.100.20'], 'data_frames':[['MA','MRL',64],['MRR','MB',63]], 'ip_endpoints':['192.0.2.10','198.51.100.20'], 'request_reply_data_counts':[2,2,2], 'receive_cases':arp_cases, 'caches':{k:{str(a):b for a,b in c.items()} for k,c in caches.items()}} # Every four-bridge graph, each undirected edge absent or cost 1/2. tree_graphs, tree_components = 0, 0 pairs = list(itertools.combinations(range(4),2)) for values in itertools.product((0,1,2),repeat=len(pairs)): edges = {(a,b):c for (u,v),c in zip(pairs,values) if c for a,b in ((u,v),(v,u))} distances = floyd(4,edges) states, active, designated, rounds = bridge_tree(4,edges) roots = set() for u,(root,cost,parent) in enumerate(states): expected_root = min(v for v in range(4) if distances[u][v] < math.inf) assert root == expected_root and cost == distances[u][root] roots.add(root) if u == root: assert parent is None else: assert cost == edges[u,parent] + states[parent][1] assert states[parent][1] < cost seen = set() v = u while states[v][2] is not None: assert v not in seen seen.add(v);v=states[v][2] assert v == root assert len(active) == 4 - len(roots) assert active == {tuple(sorted((u,s[2]))) for u,s in enumerate(states) if s[2] is not None} tree_graphs += 1;tree_components += len(roots) edges = {(a,b):c for u,v,c in [(0,1,4),(0,2,4),(1,2,1)] for a,b in ((u,v),(v,u))} st, active, designated, _ = bridge_tree(3,edges) assert st == [(0,0,None),(0,4,0),(0,4,0)] and designated[1,2] == 1 assert sum(edges[u,v] for u,v in active) == 8 # Enumerate all spanning trees of the triangle; no MST algorithm is shared. costs=[] for candidate in itertools.combinations([(0,1),(0,2),(1,2)],2): costs.append(sum(edges[e] for e in candidate)) assert min(costs) == 5 del edges[0,2];del edges[2,0] assert bridge_tree(3,edges)[0][2] == (0,5,1) out['spanning_tree']={'graphs':tree_graphs,'component_roots_checked':tree_components, 'main_tree_cost':8,'minimum_tree_cost':5,'after_failure_cost_bridge3':5} # All directed three-router graphs with missing/1/3 edges, every destination. arcs=list(itertools.permutations(range(3),2));dv_checks=0 for values in itertools.product((0,1,3),repeat=6): edges={e:c for e,c in zip(arcs,values) if c} oracle=floyd(3,edges) for dest in range(3): assert sync_dv(3,edges,dest)[-1] == [oracle[u][dest] for u in range(3)] dv_checks += 1 two = distance_failure() assert [x['value'] for x in two] == list(range(3,17))+[16] poisoned = distance_failure(poison=True) assert [x['value'] for x in poisoned] == [16,16] three = distance_failure(poison=True,three=True) assert [x['value'] for x in three] == list(range(4,17))+[16,16] long_edges={(u,u+1):1 for u in range(16)} assert sync_dv(17,long_edges,16,16)[-1][0] == 16 assert floyd(17,long_edges)[0][16] == 16 # physical path exists; representation refuses it. out['distance_vector']={'graph_destination_cases':dv_checks,'two_node_values':[x['value'] for x in two], 'poisoned_two_node_values':[x['value'] for x in poisoned], 'poisoned_three_node_values':[x['value'] for x in three], 'metric16_path_represented_unreachable':True} old={0:(10,{1:1,2:4}),1:(11,{0:1,2:1}),2:(8,{0:4,1:1})} r1={u:(s,dict(ns)) for u,(s,ns) in old.items()} r2={u:(s,dict(ns)) for u,(s,ns) in old.items()} assert ls_accept(r2,1,12,{0:1}) before=[path_to(3,ls_graph(r1),0,2),path_to(3,ls_graph(r2),1,2)] assert before == [(2,(0,1,2)),(5,(1,0,2))] assert before[0][1][1] == 1 and before[1][1][1] == 0 assert ls_accept(r1,1,12,{0:1}) and not ls_accept(r1,1,11,{0:1,2:1}) after=[path_to(3,ls_graph(r1),0,2),path_to(3,ls_graph(r2),1,2)] assert after == [(4,(0,2)),(5,(1,0,2))] permutations_checked=0 records=[(1,11,{0:1,2:1}),(1,12,{0:1}),(1,10,{0:1,2:1}), (2,8,{0:4,1:1}),(2,9,{0:4})] for order in itertools.permutations(records): db={0:(10,{1:1,2:4})} maximum={} for u,s,ns in order: accepted=ls_accept(db,u,s,ns) assert accepted == (u not in maximum or s > maximum[u]) maximum[u]=max(s,maximum.get(u,-1)) assert db[1] == (12,{0:1}) and db[2] == (9,{0:4}) assert (1,2) not in ls_graph(db) permutations_checked+=1 # Duplicate and lower-version replay after acceptance is idempotent. for sequence in range(13): db={1:(12,{0:1})} assert not ls_accept(db,1,sequence,{2:1}) assert db=={1:(12,{0:1})} out['link_state']={'before_paths':before,'after_paths':after, 'message_permutations':permutations_checked,'replay_cases':13} choices=[dict(peer=65020,path=(65020,65000),pref=100,resolvable=True), dict(peer=65030,path=(65030,65040,65000),pref=200,resolvable=True), dict(peer=65021,path=(65021,65010,65000),pref=999,resolvable=True), dict(peer=65031,path=(65031,65000),pref=999,resolvable=False)] eligible=[c for c in choices if 65010 not in c['path'] and c['resolvable']] best=min(eligible,key=lambda c:(-c['pref'],len(c['path']),c['peer'])) assert best['peer']==65030 fixed=[] for bits in itertools.product((False,True),repeat=3): current={i:((i,i%3+1,0) if bits[i-1] else (i,0)) for i in (1,2,3)} if all(policy_choice(i,current)==current[i] for i in (1,2,3)): fixed.append(bits) assert not fixed # Include empty paths as well: direct choices always dominate them. empty_fixed=[] for values in itertools.product((0,1,2),repeat=3): current={i:(() if values[i-1]==0 else ((i,0) if values[i-1]==1 else (i,i%3+1,0))) for i in (1,2,3)} if all(policy_choice(i,current)==current[i] for i in (1,2,3)): empty_fixed.append(values) assert not empty_fixed current={i:(i,0) for i in (1,2,3)} current[1]=policy_choice(1,current) initial=dict(current);cycle=[] for i in [2,1,3,2,1,3]: current[i]=policy_choice(i,current) assert all(len(p)==len(set(p)) for p in current.values()) cycle.append(''.join('D' if len(current[j])==2 else 'I' for j in (1,2,3))) assert cycle == ['IID','DID','DII','DDI','IDI','IDD'] and current==initial # Real FIFO channels: receive one preferred-neighbor message per shown event. # Other-neighbor messages are drained at each activation and cannot supply an allowed I path. fifo_current = {i:(i,0) for i in (1,2,3)} received = {i:{} for i in (1,2,3)} channels = {(u,v):deque([fifo_current[u]]) for u in (1,2,3) for v in (1,2,3) if u != v} deliveries = 0 def receive(u,v): nonlocal deliveries received[v][u] = channels[u,v].popleft() deliveries += 1 successor = v % 3 + 1 chosen = (v,successor,0) if received[v].get(successor) == (successor,0) else (v,0) if chosen != fifo_current[v]: fifo_current[v] = chosen for w in (1,2,3): if w != v: channels[v,w].append(chosen) def activate(v): successor = v % 3 + 1 for u in (1,2,3): if u not in (v,successor): while channels[u,v]: receive(u,v) assert channels[successor,v] receive(successor,v) return ''.join('D' if len(fifo_current[j])==2 else 'I' for j in (1,2,3)) assert activate(1) == 'IDD' for _ in range(100): assert [activate(v) for v in [2,1,3,2,1,3]] == cycle assert max(map(len,channels.values())) <= 2 fifo_deliveries = deliveries # A changed policy (all prefer direct) has the advertised stable assignment. all_direct={i:(i,0) for i in (1,2,3)} assert all(policy_choice(i,all_direct,False)==all_direct[i] for i in (1,2,3)) out['path_vector']={'chosen_peer':best['peer'],'exported_path':[65010,*best['path']], 'binary_assignments':8,'including_empty_assignments':27,'stable_assignments':0, 'fair_cycle':cycle,'fifo_cycles_checked':100,'fifo_deliveries':fifo_deliveries, 'all_direct_policy_is_stable':True} print(json.dumps(out,ensure_ascii=False,indent=2,allow_nan=False)) if __name__ == '__main__': main()