145 lines
5.5 KiB
Python
145 lines
5.5 KiB
Python
"""Build the lab network, apply your plan, and check it. Given; do not change it.
|
|
|
|
Run inside the course container, from this folder:
|
|
|
|
python3 lab.py check # build, apply plan.py, run the checks, stop
|
|
python3 lab.py cli # build, apply plan.py, then a mininet> prompt
|
|
|
|
The network, three LANs cut from 192.168.1.0/24:
|
|
|
|
net A 192.168.1.0/26 net B 192.168.1.64/26 net C 192.168.1.128/26
|
|
h1 --+ +-- h3
|
|
s1 -- r1 ------------- s2 ---------- r2 -- s3 -- h4
|
|
h2 --+ (r1-eth0 | r1-eth1) (r2-eth0 | r2-eth1)
|
|
|
|
Every interface starts with no address. plan.py says what to run on each node.
|
|
Mininet reference: http://mininet.org/walkthrough/
|
|
"""
|
|
|
|
import ipaddress
|
|
import sys
|
|
|
|
from mininet.log import setLogLevel
|
|
from mininet.net import Mininet
|
|
from mininet.nodelib import LinuxBridge
|
|
from mininet.topo import Topo
|
|
|
|
import plan
|
|
|
|
NETS = {"A": ipaddress.ip_network("192.168.1.0/26"),
|
|
"B": ipaddress.ip_network("192.168.1.64/26"),
|
|
"C": ipaddress.ip_network("192.168.1.128/26")}
|
|
# which network each interface is on
|
|
INTERFACES = {"h1-eth0": "A", "h2-eth0": "A", "r1-eth0": "A",
|
|
"r1-eth1": "B", "h3-eth0": "B", "r2-eth0": "B",
|
|
"r2-eth1": "C", "h4-eth0": "C"}
|
|
HOSTS = ["h1", "h2", "h3", "h4"]
|
|
|
|
|
|
class Lab(Topo):
|
|
def build(self):
|
|
s1, s2, s3 = (self.addSwitch(s) for s in ("s1", "s2", "s3"))
|
|
for name in ("h1", "h2", "h3", "h4", "r1", "r2"):
|
|
self.addHost(name, ip=None)
|
|
self.addLink("h1", s1, intfName1="h1-eth0")
|
|
self.addLink("h2", s1, intfName1="h2-eth0")
|
|
self.addLink("r1", s1, intfName1="r1-eth0")
|
|
self.addLink("r1", s2, intfName1="r1-eth1")
|
|
self.addLink("h3", s2, intfName1="h3-eth0")
|
|
self.addLink("r2", s2, intfName1="r2-eth0")
|
|
self.addLink("r2", s3, intfName1="r2-eth1")
|
|
self.addLink("h4", s3, intfName1="h4-eth0")
|
|
|
|
|
|
def start():
|
|
net = Mininet(topo=Lab(), switch=LinuxBridge, controller=None, autoSetMacs=True)
|
|
net.start()
|
|
for node in net.hosts:
|
|
for intf in node.intfList():
|
|
if intf.name != "lo":
|
|
node.cmd(f"ip addr flush dev {intf.name}")
|
|
node.cmd("sysctl -qw net.ipv4.ip_forward=0")
|
|
for node, commands in plan.PLAN.items():
|
|
for command in commands:
|
|
out = net.get(node).cmd(command).strip()
|
|
if out:
|
|
print(f"{node}$ {command}\n {out}")
|
|
return net
|
|
|
|
|
|
def addresses(net):
|
|
"""{interface: IPv4 interface address} as configured."""
|
|
found = {}
|
|
for name in INTERFACES:
|
|
node = net.get(name.split("-")[0])
|
|
for line in node.cmd(f"ip -4 -o addr show dev {name}").splitlines():
|
|
found[name] = ipaddress.ip_interface(line.split()[3])
|
|
return found
|
|
|
|
|
|
def ip_of(net, host):
|
|
found = addresses(net).get(f"{host}-eth0")
|
|
if found is None:
|
|
raise LookupError(f"{host} has no address yet, so this cannot be tested (see the first check)")
|
|
return str(found.ip)
|
|
|
|
|
|
def check_addresses(net):
|
|
found, problems = addresses(net), []
|
|
for name, where in INTERFACES.items():
|
|
a = found.get(name)
|
|
if a is None:
|
|
problems.append(f"{name} has no IPv4 address")
|
|
elif a.network != NETS[where]:
|
|
problems.append(f"{name} is {a}, but it is on net {where}, {NETS[where]}")
|
|
elif a.ip in (NETS[where].network_address, NETS[where].broadcast_address):
|
|
problems.append(f"{name} is {a.ip}, which is net {where}'s network or broadcast address")
|
|
return "; ".join(problems) or None
|
|
|
|
|
|
def check_reachability(net):
|
|
missing = []
|
|
for a in HOSTS:
|
|
for b in HOSTS:
|
|
# up to three tries: the first packet on a new network can be lost while it settles
|
|
if a != b and " 0 received" in net.get(a).cmd(f"ping -c 3 -i 0.3 -W 1 {ip_of(net, b)}"):
|
|
missing.append(f"{a}->{b}")
|
|
return f"no reply for {', '.join(missing)}" if missing else None
|
|
|
|
|
|
def check_path(net):
|
|
out = net.get("h1").cmd(f"traceroute -n -w 1 -q 1 {ip_of(net, 'h4')}")
|
|
hops = [line.split()[1] for line in out.splitlines()[1:] if line.strip() and line.split()[1] != "*"]
|
|
if len(hops) != 3:
|
|
return f"h1 to h4 should cross r1 and r2 (three hops), traceroute shows {hops}"
|
|
ttl = [w for w in net.get("h1").cmd(f"ping -c 1 {ip_of(net, 'h4')}").split() if w.startswith("ttl=")]
|
|
return None if ttl == ["ttl=62"] else f"h4's reply should arrive with TTL 62 (two routers), got {ttl}"
|
|
|
|
|
|
def main():
|
|
setLogLevel("info")
|
|
mode = sys.argv[1] if len(sys.argv) > 1 else "check"
|
|
net = start()
|
|
try:
|
|
if mode == "cli":
|
|
from mininet.cli import CLI
|
|
CLI(net)
|
|
return
|
|
results = []
|
|
for title, check in [("every interface is on its own network", check_addresses),
|
|
("every host reaches every other host", check_reachability),
|
|
("h1 to h4 crosses both routers", check_path)]:
|
|
try:
|
|
problem = check(net)
|
|
except Exception as e: # reported, then the next check runs
|
|
problem = str(e) if isinstance(e, LookupError) else f"{type(e).__name__}: {e}"
|
|
results.append(problem is None)
|
|
print(("PASS " if problem is None else "FAIL ") + title + ("" if problem is None else f": {problem}"))
|
|
print(f"\n{sum(results)} passed, {len(results) - sum(results)} failed")
|
|
sys.exit(0 if all(results) else 1)
|
|
finally:
|
|
net.stop()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|