Hands-on 04-ports-protocols-and-host-firewalls: Lab

safety_constraints

Do not change the host's real firewall configuration.
Do not bind either listener to a wildcard address or an externally reachable interface.
Create and modify files only below /opt/lab-classroom/class25/.
Use only unprivileged ports.
Run the listener as the current non-root laboratory user.
The policy evaluator is a simulator and must not be represented as active enforcement.

workspace

/opt/lab-classroom/class25/

files

path

/opt/lab-classroom/class25/listener.py

content

import socket
import sys

if len(sys.argv) != 3 or sys.argv[1] not in {"tcp", "udp"}:
raise SystemExit("usage: python3 listener.py tcp|udp PORT")

protocol = sys.argv[1]
port = int(sys.argv[2])
if not 1024 <= port <= 65535:
raise SystemExit("use an unprivileged port from 1024 through 65535")

kind = socket.SOCK_STREAM if protocol == "tcp" else socket.SOCK_DGRAM
with socket.socket(socket.AF_INET, kind) as server:
server.bind(("127.0.0.1", port))
if protocol == "tcp":
server.listen(1)
print(f"READY {protocol} 127.0.0.1:{port}", flush=True)
if protocol == "tcp":
connection, peer = server.accept()
with connection:
data = connection.recv(1024)
connection.sendall(b"ACK:" + data)
print(f"TCP_PEER {peer[0]}:{peer[1]} DATA {data.decode(errors='replace')}", flush=True)
else:
data, peer = server.recvfrom(1024)
server.sendto(b"ACK:" + data, peer)
print(f"UDP_PEER {peer[0]}:{peer[1]} DATA {data.decode(errors='replace')}", flush=True)

path

/opt/lab-classroom/class25/probe.py

content

import socket
import sys

if len(sys.argv) != 5 or sys.argv[1] not in {"tcp", "udp"}:
raise SystemExit("usage: python3 probe.py tcp|udp HOST PORT MESSAGE")

protocol, host, port_text, message = sys.argv[1:]
port = int(port_text)
kind = socket.SOCK_STREAM if protocol == "tcp" else socket.SOCK_DGRAM
with socket.socket(socket.AF_INET, kind) as client:
client.settimeout(3)
if protocol == "tcp":
client.connect((host, port))
client.sendall(message.encode())
reply = client.recv(1024)
else:
client.sendto(message.encode(), (host, port))
reply, _ = client.recvfrom(1024)
print(reply.decode(errors="replace"))

path

/opt/lab-classroom/class25/policy.json

content

{
"defaults": {
"ingress": "DROP",
"egress": "ACCEPT"
},
"rules": [
{
"name": "allow-established",
"direction": "ingress",
"states": ["established", "related"],
"action": "ACCEPT"
},
{
"name": "allow-local-web",
"direction": "ingress",
"protocol": "tcp",
"source": "127.0.0.0/8",
"destination_port": 18080,
"states": ["new"],
"action": "ACCEPT"
},
{
"name": "allow-local-dns",
"direction": "ingress",
"protocol": "udp",
"source": "127.0.0.0/8",
"destination_port": 15353,
"states": ["new"],
"action": "ACCEPT"
}
]
}

path

/opt/lab-classroom/class25/evaluate.py

content

import ipaddress
import json
from pathlib import Path

root = Path("/opt/lab-classroom/class25")
policy = json.loads((root / "policy.json").read_text())
packets = [
{"name": "established-reply", "direction": "ingress", "protocol": "tcp", "source": "192.0.2.10", "destination_port": 49152, "state": "established"},
{"name": "local-web", "direction": "ingress", "protocol": "tcp", "source": "127.0.0.1", "destination_port": 18080, "state": "new"},
{"name": "remote-web", "direction": "ingress", "protocol": "tcp", "source": "192.0.2.10", "destination_port": 18080, "state": "new"},
{"name": "local-dns", "direction": "ingress", "protocol": "udp", "source": "127.0.0.1", "destination_port": 15353, "state": "new"},
{"name": "outbound-query", "direction": "egress", "protocol": "udp", "source": "192.0.2.20", "destination_port": 53, "state": "new"}
]

def matches(rule, packet):
if rule.get("direction") != packet["direction"]:
return False
if "protocol" in rule and rule["protocol"] != packet["protocol"]:
return False
if "destination_port" in rule and rule["destination_port"] != packet["destination_port"]:
return False
if "states" in rule and packet["state"] not in rule["states"]:
return False
if "source" in rule:
address = ipaddress.ip_address(packet["source"])
if address not in ipaddress.ip_network(rule["source"]):
return False
return True

for packet in packets:
for rule in policy["rules"]:
if matches(rule, packet):
print(packet["name"], rule["action"], rule["name"])
break
else:
action = policy["defaults"][packet["direction"]]
print(packet["name"], action, "default-" + packet["direction"])

procedure

Confirm that /opt/lab-classroom/class25/ already exists, is writable by the laboratory user, and is the current working directory.
Create listener.py, probe.py, policy.json, and evaluate.py at the exact listed paths using the supplied contents. Do not create laboratory files elsewhere.
Start the TCP listener with: python3 listener.py tcp 18080 > tcp.log 2>&1 & echo $! > tcp.pid
Wait until tcp.log contains a line beginning with READY. If it does not appear within five seconds, stop and troubleshoot rather than repeatedly starting listeners.
Run the TCP client with: python3 probe.py tcp 127.0.0.1 18080 hello-tcp
Confirm that the client prints ACK:hello-tcp and that the listener exits after serving the connection.
Start the UDP listener with: python3 listener.py udp 15353 > udp.log 2>&1 & echo $! > udp.pid
Wait until udp.log contains a line beginning with READY.
Run the UDP client with: python3 probe.py udp 127.0.0.1 15353 hello-udp
Confirm that the client prints ACK:hello-udp and that the listener exits after processing the datagram.
Run the simulated policy evaluation with: python3 evaluate.py | tee evaluation.log
Compare each decision with the ordered policy. Explain why remote-web reaches the ingress default while local-web matches an explicit rule.
Edit only /opt/lab-classroom/class25/policy.json and temporarily change the allow-local-web destination_port from 18080 to 18081.
Run python3 evaluate.py again and observe that local-web is now dropped by the default ingress policy.
Restore destination_port to 18080 and rerun the evaluator to confirm the original result.