""" Trace Autonomous System (AS) paths from source to destination Author: Tim Pierson, Dartmouth CS60, Fall 2026 Developed with Claude Code Opus 5.5 Medium First run traceroute and save the results (-n prints numeric IP addresses instead of looking up hostnames): traceroute -n 8.8.8.8 > google.txt traceroute -n 1.1.1.1 > cloudflare.txt traceroute -n dartmouth.edu > campus.txt traceroute -n www.u-tokyo.ac.jp > japan.txt traceroute -n www.apple.com > apple.txt traceroute -n www.ox.ac.uk > europe.txt Next run this code to see which ASes were crossed: python3 as_explorer.py Notes: Reads every .txt file in the current directory as one trace, so keep only your traceroute output files there. One traceroute tells you almost nothing about inter-domain routing. It is a list of routers. The AS-level structure only appears when you compare paths to DIFFERENT destinations and ask what they share. That is roughly how researchers infer the shape of the Internet: nobody publishes the AS graph, so you measure many paths and look at the overlap. No root needed. Reads trace files, looks the hops up in the global BGP table via Team Cymru (falling back to the RADb routing registry for router addresses that are not in BGP) """ import glob import re import socket import sys from collections import Counter CYMRU_HOST = "whois.cymru.com" RADB_HOST = "whois.radb.net" WHOIS_PORT = 43 # =========================================================================== # Part 1: from a router path to an AS path # =========================================================================== def as_path(hops): """ Reduce a list of hops to the sequence of ASes the packet crossed. `hops` is a list of dicts with keys 'num', 'ips', 'asn'. 'asn' is an integer, or None for a private address or a hop that did not reply. Three things to do, and each one is a statement about routing: 1. Drop hops with no ASN. A private address is not in anybody's BGP table, and a silent router tells you nothing. Neither is a step in the AS path. 2. Collapse consecutive runs of the same ASN into one entry. Four hops inside one AS are four forwarding decisions but ONE step in the AS path -- the AS path records which organizations were crossed, not how many routers each one used. 3. Keep the order. Return a list of ints, e.g. [10755, 3999, 6509, 15169]. Careful: do not collapse non-adjacent repeats. If a path genuinely leaves an AS and comes back, [1, 2, 1] is three steps, not two. """ path = [] for hop in hops: if hop["asn"] is None: continue # no AS: not a step in the AS path if path and path[-1] == hop["asn"]: continue # same AS as the previous step path.append(hop["asn"]) return path # =========================================================================== # Part 2: how many ways out of your network are there? # =========================================================================== def common_prefix(paths): """ Return the longest sequence of ASes that EVERY path begins with. `paths` is a list of AS paths (each a list of ints). Why this matters: if every path to every destination starts with the same two or three ASes, your network has that many ways to reach the rest of the Internet -- and often it is exactly one. The whole global routing table collapses, from your vantage point, into a single first move. Stop at the first position where the paths disagree, or where any path runs out. """ if not paths: return [] shared = [] for position in range(min(len(p) for p in paths)): here = paths[0][position] if all(p[position] == here for p in paths): shared.append(here) else: break # first disagreement ends the prefix return shared # =========================================================================== # Part 3: where does your traffic get sorted by destination? # =========================================================================== def divergence(path_a, path_b): """ Return the last AS that both paths share before they split, or None if they never agreed at all. This AS is where a routing decision actually distinguished your two destinations. Everything before it treated both packets identically. Everything after is a different set of organizations. That AS is doing inter-domain policy: it holds BGP routes for both destinations and chose a different next hop for each. """ last = None for a, b in zip(path_a, path_b): if a != b: break last = a return last # =========================================================================== # Part 4: what are you most dependent on? # =========================================================================== def transit_dependency(paths, home_asn): """ Return a list of (asn, count) sorted by how many paths cross that AS, most first, excluding your own AS. Read the top entry as: if that single organization went down, this many of your destinations become unreachable. Your own AS is excluded because it appears in everything by definition and tells you nothing. An AS crossed by every path is a chokepoint. Notice that it need not be large or famous -- it just has to be the only way out. """ counts = Counter() for path in paths: for asn in set(path): # set(): count each path once if asn != home_asn: counts[asn] += 1 return sorted(counts.items(), key=lambda kv: (-kv[1], kv[0])) # =========================================================================== # Part 5: the AS-level graph # =========================================================================== def as_graph(paths): """ Build the AS adjacency graph these paths reveal. Return a dict mapping each ASN to the set of ASNs it was seen directly connected to. Every consecutive pair in every path is an edge, and the edge is undirected -- observing A then B tells you A and B are neighbours. This is a different graph from the one in the Dijkstra slides. There, nodes were routers inside a single network and the operator knew every link. Here nodes are whole organizations, nobody publishes the edges, and you are inferring them from measurement. """ graph = {} for path in paths: for a, b in zip(path, path[1:]): graph.setdefault(a, set()).add(b) graph.setdefault(b, set()).add(a) return graph def degrees(graph): """ Return [(asn, degree)] sorted by degree, highest first. Your own AS should come out near the bottom: a campus network connects to one or two upstreams. The AS at the top is carrying traffic between many others, which is what being in the middle of the hierarchy looks like from the outside. """ return sorted(((asn, len(nbrs)) for asn, nbrs in graph.items()), key=lambda kv: (-kv[1], kv[0])) # =========================================================================== # Provided: parsing, address classification, AS lookup # =========================================================================== IP_RE = re.compile(r"\b(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})\b") PRIVATE_BLOCKS = [ ("10.0.0.0", 8), ("172.16.0.0", 12), ("192.168.0.0", 16), ("127.0.0.0", 8), ("169.254.0.0", 16), ("100.64.0.0", 10), ] def ip_to_int(dotted): a, b, c, d = (int(x) for x in dotted.split(".")) return (a << 24) | (b << 16) | (c << 8) | d def in_block(ip, network, prefix_len): mask = (0xFFFFFFFF << (32 - prefix_len)) & 0xFFFFFFFF return (ip_to_int(ip) & mask) == (ip_to_int(network) & mask) def is_private(ip): """True if this address is not in the global BGP table.""" return any(in_block(ip, n, p) for n, p in PRIVATE_BLOCKS) def parse_traceroute(text): """Pull hop numbers and IPs out of `traceroute -n` output.""" hops = [] for line in text.splitlines(): line = line.rstrip() if not line or line.lstrip().startswith("traceroute to"): continue m = re.match(r"\s*(\d+)\s+(.*)", line) if not m: continue num, rest = int(m.group(1)), m.group(2) rest = re.sub(r"\d+\.\d+\s*ms", "", rest) ips = [] for ip in IP_RE.findall(rest): if ip not in ips: ips.append(ip) hops.append({"num": num, "ips": ips, "asn": None, "name": None}) return hops def whois(host, query): """Send one query to a whois server on port 43 and return the reply.""" with socket.create_connection((host, WHOIS_PORT), timeout=20) as s: s.sendall(query.encode()) chunks = [] while True: data = s.recv(8192) if not data: break chunks.append(data) return b"".join(chunks).decode(errors="replace") def as_name(field): """ Tidy Team Cymru's AS name field for display. Cymru gives 'HANDLE - Organization, CC', e.g. 'TVC-AS1 - FirstLight Fiber, Inc., US'. The handle was picked when the AS was registered and can outlive the company (TVC-AS1 is now run by FirstLight), so keep both and drop only the trailing country code. """ return re.sub(r",\s*[A-Z]{2}$", "", field.strip()) def lookup_asns(ips): """ Look up many addresses in ONE whois connection to Team Cymru. Bulk format: 'begin', 'verbose', one address per line, 'end'. Returns {ip: (asn, as_name)}; unrouted addresses are left out. """ if not ips: return {} try: raw = whois(CYMRU_HOST, "begin\nverbose\n" + "\n".join(ips) + "\nend\n") except OSError as e: print(f" [AS lookup failed: {e}]") print(f" [is outbound port 43 open?]") return {} results = {} for line in raw.splitlines(): if "|" not in line or line.startswith("Bulk mode"): continue parts = [p.strip() for p in line.split("|")] if len(parts) < 7 or parts[0] == "AS": continue if not parts[0].isdigit(): continue results[parts[1]] = (int(parts[0]), as_name(parts[6])) return results def lookup_as_names(asns): """Ask Team Cymru for the names of these ASNs. Returns {asn: name}.""" if not asns: return {} query = "begin\nverbose\n" + "\n".join(f"AS{a}" for a in asns) + "\nend\n" try: raw = whois(CYMRU_HOST, query) except OSError: return {} results = {} for line in raw.splitlines(): parts = [p.strip() for p in line.split("|")] if len(parts) == 5 and parts[0].isdigit(): results[int(parts[0])] = as_name(parts[4]) return results def lookup_registry(ips): """ Fallback for addresses Team Cymru could not place: ask the RADb routing registry which AS is REGISTERED to originate each one. Cymru answers from live BGP. Some networks (Internet2 is one) never announce the addresses of their own routers to the commercial Internet, so those hops have no BGP route at all. The registry still records who holds the block -- weaker evidence than BGP, since it says who SHOULD originate the prefix, not what routers are actually doing. Only RPKI route objects are used. Those are cryptographically signed by whoever holds the address block. Other registry entries are typed in by hand and are often stale -- an exchange-point address on your path can match an old entry for a company that has nothing to do with it. Several route objects usually cover one address. Use longest-prefix match, the same rule a router uses. If the most specific prefix lists more than one origin, prefer the AS that also signed a covering, less specific block (the block's holder); otherwise the lowest ASN. All queries go over one connection in IRRd's persistent mode ('!!'); '!r/32,L' asks for every route object covering the address. Returns {ip: asn}; addresses with no RPKI route object are left out. """ if not ips: return {} query = "!!\n" + "".join(f"!r{ip}/32,L\n" for ip in ips) + "!q\n" try: raw = whois(RADB_HOST, query).encode() except OSError: return {} # Each answer is 'A' + that many bytes + 'C', or a one-line # 'C' / 'D' / 'F ...' when there is nothing to return. They arrive # in the same order as the queries. answers = [] pos = 0 while pos < len(raw) and len(answers) < len(ips): end = raw.find(b"\n", pos) if end == -1: break head = raw[pos:end].strip() pos = end + 1 if head.startswith(b"A"): length = int(head[1:]) answers.append(raw[pos:pos + length].decode(errors="replace")) pos += length elif head[:1] in (b"D", b"F"): answers.append("") # anything else ('C' closing an A block, blank lines) is framing results = {} for ip, text in zip(ips, answers): routes = [] # (prefix_len, asn) for obj in text.split("\n\n"): fields = dict(line.split(":", 1) for line in obj.splitlines() if ":" in line and not line.startswith(" ")) if not fields.get("source", "").strip().startswith("RPKI"): continue net, length = fields["route"].strip().split("/") asn = fields["origin"].strip().upper().removeprefix("AS") if asn.isdigit() and in_block(ip, net, int(length)): routes.append((int(length), int(asn))) if not routes: continue longest = max(length for length, _ in routes) candidates = sorted({asn for length, asn in routes if length == longest}) holders = [a for a in candidates if any(asn == a for length, asn in routes if length < longest)] results[ip] = (holders or candidates)[0] return results def annotate_all(traces): """ Resolve every public address across every trace: one bulk lookup to Team Cymru, then the routing registry for anything Cymru could not place. ASes known only from the registry are labelled '[registry]'. """ public = [] for hops in traces.values(): for hop in hops: for ip in hop["ips"]: if not is_private(ip) and ip not in public: public.append(ip) table = lookup_asns(public) missing = [ip for ip in public if ip not in table] if missing: print(f" [{len(missing)} addresses not in BGP; checking the routing" f" registry...]") from_registry = lookup_registry(missing) known = {asn for asn, _ in table.values()} new_asns = sorted(set(from_registry.values()) - known) as_names = lookup_as_names(new_asns) for ip, asn in from_registry.items(): name = as_names.get(asn, "?") if asn not in known: name += " [registry]" table[ip] = (asn, name) names = {} for hops in traces.values(): for hop in hops: for ip in hop["ips"]: if ip in table: hop["asn"], hop["name"] = table[ip] names[hop["asn"]] = hop["name"] break return names # =========================================================================== # Provided: the AS tree # =========================================================================== # Box-drawing characters. If a terminal ever mangles these, fall back to # BRANCH, LAST, PIPE, BLANK = "|-- ", "`-- ", "| ", " " BRANCH, LAST, PIPE, BLANK = "├── ", "└── ", "│ ", " " def build_tree(paths): """ Build a trie of the AS paths. Because every path starts at your own AS, the paths share a root and fan out. The trunk is the stretch where destinations have no choice; each fork is an AS that sorted your traffic by destination. Each node is a dict: {'asn': int, 'children': {asn: node}, 'dests': [names that END here], 'count': int} 'count' is how many paths pass through the node. """ root = {"asn": None, "children": {}, "dests": [], "count": 0} for dest, path in paths.items(): if not path: continue node = root for asn in path: node = node["children"].setdefault( asn, {"asn": asn, "children": {}, "dests": [], "count": 0}) node["count"] += 1 node["dests"].append(dest) return root def print_as_tree(paths, names, home_asn): """ Draw the trie, marking the shared trunk and every shared dependency. What to read off it: * the trunk, before the first fork, is forced -- no destination has an alternative * a node's [n paths] is how many destinations die if that single organization goes down * a fork is an AS holding BGP routes for several of your destinations and choosing differently for each """ root = build_tree(paths) if not root["children"]: return total = len([p for p in paths.values() if p]) # An ASN can appear in more than one branch (left one AS, came back by # a different route). The trie shows it twice, so flag it. seen_nodes = Counter() def tally(node): for child in node["children"].values(): seen_nodes[child["asn"]] += 1 tally(child) tally(root) print("\n" + "=" * 72) print("The AS tree: where your paths agree, and where they split") print("=" * 72) def draw(node, prefix, is_last): connector = LAST if is_last else BRANCH tags = [] if node["asn"] == home_asn: tags.append("<-- your AS") if node["count"] > 1: tags.append(f"[{node['count']} paths]") if (node["count"] == total and total > 1 and node["asn"] != home_asn): tags.append("<-- EVERY path") if seen_nodes[node["asn"]] > 1: tags.append("(also on another branch)") for d in node["dests"]: tags.append(f"-> {d}") print(f" {prefix}{connector}{label(node['asn'], names)}" + (" " + " ".join(tags) if tags else "")) kids = sorted(node["children"].values(), key=lambda n: (-n["count"], n["asn"])) for i, kid in enumerate(kids): draw(kid, prefix + (BLANK if is_last else PIPE), i == len(kids) - 1) # You are the root; every path's first AS hangs off you print(" You") for i, kid in enumerate(sorted(root["children"].values(), key=lambda n: (-n["count"], n["asn"]))): draw(kid, "", i == len(root["children"]) - 1) # Summarise the trunk in words, since the shape alone is easy to skim past trunk = [] node = root while len(node["children"]) == 1: node = next(iter(node["children"].values())) trunk.append(node["asn"]) # Only meaningful if the trunk actually ends in a fork. With one # trace there is no choice anywhere, so say nothing. if len(trunk) > 1 and len(node["children"]) > 1: print(f"\n Trunk: " + " -> ".join(f"AS{a}" for a in trunk)) print(f" {len(trunk)} ASes every destination must cross. The first" f" real choice happens at") print(f" AS{trunk[-1]}, which is where your traffic starts being" f" sorted by destination.") # =========================================================================== # Provided: report # =========================================================================== def label(asn, names): return f"AS{asn} ({names.get(asn, '?')})" def report(paths, names, home_asn): print("\n" + "=" * 72) print("AS-PATH per destination") print("=" * 72) for dest, path in paths.items(): if not path: print(f" {dest:<16} (no AS path)") continue print(f" {dest:<16} " + " -> ".join(f"AS{a}" for a in path)) if not any(paths.values()): return print_as_tree(paths, names, home_asn) print("\n" + "=" * 72) print("1. How many ways out of your network?") print("=" * 72) external = [p for d, p in paths.items() if p and p != [home_asn]] shared = common_prefix(external) # The AS right after yours on each path is an upstream you buy # connectivity from (or peer with) exits = sorted({p[1] for p in external if len(p) > 1 and p[0] == home_asn}) if not exits: print(" (none)") elif len(exits) == 1: print(" Every path off your network begins:") print(" " + " -> ".join(label(a, names) for a in shared)) print(f"\n {len(shared)} ASes of forced agreement. From your network," f" the global routing") print(f" table offers exactly one first move.") else: print(f" Your traffic leaves through {len(exits)} different upstream" f" networks:") for a in exits: dests = [d for d, p in paths.items() if len(p) > 1 and p[1] == a] print(f" {label(a, names)} -> {', '.join(dests)}") print("\n Your border routers hear routes from each one and use BGP" " to pick,") print(" destination by destination, which way out to use.") print("\n" + "=" * 72) print("2. Where does your traffic get sorted by destination?") print("=" * 72) dests = [d for d, p in paths.items() if p] seen = set() for i, a in enumerate(dests): for b in dests[i + 1:]: d = divergence(paths[a], paths[b]) if d is None: continue seen.add(d) print(f" {a:<14} vs {b:<14} split after {label(d, names)}") if seen: print(f"\n {len(seen)} AS(es) make the decisions that separate your" f" destinations.") print("\n" + "=" * 72) print("3. What are you most dependent on?") print("=" * 72) deps = transit_dependency(list(paths.values()), home_asn) total = len([p for p in paths.values() if p]) if deps: width = max(len(label(asn, names)) for asn, _ in deps[:6]) for asn, count in deps[:6]: bar = "#" * count flag = " <-- chokepoint" if count == total else "" print(f" {label(asn, names):<{width}} {count}/{total} paths {bar}{flag}") else: print(" (none)") graph = as_graph(list(paths.values())) if graph: print("\n" + "=" * 72) print("4. The AS graph your measurements reveal") print("=" * 72) top = degrees(graph)[:8] width = max(len(label(asn, names)) for asn, _ in top) for asn, deg in top: neighbours = ", ".join(f"AS{n}" for n in sorted(graph[asn])) mine = " (you)" if asn == home_asn else "" print(f" {label(asn, names):<{width}} degree {deg}{mine}") print(f" {'':<{width}} -- {neighbours}") if __name__ == "__main__": traces = {} for path in sorted(glob.glob("*.txt")): name = path.replace(".txt", "").replace("trace_", "") try: traces[name] = parse_traceroute(open(path).read()) except OSError as e: print(f"skipping {path}: {e}") if not traces: sys.exit("No .txt trace files found in the current directory.") names = annotate_all(traces) paths = {dest: as_path(hops) for dest, hops in traces.items()} # Your own AS is the first one on any path home = None for p in paths.values(): if p: home = p[0] break print(f"Read {len(traces)} traces. Your AS appears to be " f"{label(home, names) if home else 'unknown'}.") report(paths, names, home)