Working with UDP on localhost

This commit is contained in:
John Burwell 2025-04-26 22:28:52 -05:00
parent 6599484af0
commit 258b59198a
4 changed files with 240 additions and 0 deletions

32
.gitignore vendored Normal file
View File

@ -0,0 +1,32 @@
# Python artifacts
__pycache__/
*.pyc
*.pyo
*.pyd
# Virtual environment
.venv/
env/
venv/
# OS junk
.DS_Store
Thumbs.db
# Testing and coverage
.coverage
pytest_cache/
htmlcov/
# IDE/editor stuff
.vscode/
.idea/
*.swp
# Logs (if you decide to generate any)
*.log
logs/
# Radio mock dumps (if you log traffic to files later)
*.cap
*.pcap

94
base_station/server.py Normal file
View File

@ -0,0 +1,94 @@
import socket
import time
import threading
BROADCAST_IP = "127.0.0.1"
PORT = 4444
BEACON_INTERVAL = 10 # seconds
BASE_CALLSIGN = "KI5QKX-10"
NETWORK_NAME = "HAMNET-HOUSTON"
# Fake IP pool
IP_POOL = [
"44.127.254.12",
"44.127.254.13",
"44.127.254.14",
"44.127.254.15"
]
LEASE_TIME = 3600 # seconds
NETMASK = "24"
GATEWAY = "44.127.254.1"
DNS_SERVER = "44.127.254.1"
active_leases = {}
def build_beacon():
return f"0.1|CRAP_BEACON|{BASE_CALLSIGN}|{NETWORK_NAME}"
def build_accept(client_callsign, ip_address):
return f"0.1|CRAP_ACCEPT|{client_callsign}|{NETWORK_NAME}|{ip_address}/{NETMASK}|{GATEWAY}|{DNS_SERVER}|{LEASE_TIME}"
def handle_request(data, addr, sock):
parts = data.split('|')
if parts[1] == "CRAP_BEACON":
return
if len(parts) < 4:
print(f"Received malformed message from {addr}: {data}")
return
if parts[1] != "CRAP_REQUEST":
print(f"Ignoring non-request message: {data}")
return
client_callsign = parts[2]
network_requested = parts[3]
if network_requested != NETWORK_NAME:
print(f"Client {client_callsign} requested wrong network ({network_requested}). Ignored.")
return
print(f"Received join request from {client_callsign} ({addr}).")
# Assign IP
if client_callsign in active_leases:
assigned_ip = active_leases[client_callsign]
print(f"Client {client_callsign} already has IP {assigned_ip}.")
elif IP_POOL:
assigned_ip = IP_POOL.pop(0)
active_leases[client_callsign] = assigned_ip
print(f"Assigned IP {assigned_ip} to {client_callsign}.")
else:
print(f"No available IPs for {client_callsign}!")
return
# Send ACCEPT
accept_message = build_accept(client_callsign, assigned_ip)
sock.sendto(accept_message.encode('utf-8'), addr)
print(f"Sent CRAP_ACCEPT to {client_callsign}.")
def beacon_loop(sock):
while True:
beacon = build_beacon().encode('utf-8')
sock.sendto(beacon, (BROADCAST_IP, PORT))
print(f"Sent beacon: {beacon.decode('utf-8')}")
time.sleep(BEACON_INTERVAL)
def main():
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
# sock.bind((BROADCAST_IP, PORT))
print(f"Base Station {BASE_CALLSIGN} starting up on {NETWORK_NAME}...")
# Start beaconing in a separate thread
threading.Thread(target=beacon_loop, args=(sock,), daemon=True).start()
while True:
data, addr = sock.recvfrom(1024)
decoded = data.decode('utf-8')
handle_request(decoded, addr, sock)
if __name__ == "__main__":
main()

85
client_node/client.py Normal file
View File

@ -0,0 +1,85 @@
import socket
import time
LISTEN_IP = "0.0.0.0"
PORT = 4444
CLIENT_CALLSIGN = "N0CALL-7"
configured = False
lease_expiration = None
def parse_message(data):
parts = data.split('|')
return parts
def build_request(network_name):
return f"0.1|CRAP_REQUEST|{CLIENT_CALLSIGN}|{network_name}"
def apply_network_config(assigned_ip, gateway, dns, lease_time):
global configured, lease_expiration
print("\n[Network Configuration]")
print(f" Bringing up interface: ax0")
print(f" Assigned IP address: {assigned_ip}")
print(f" Default Gateway: {gateway}")
print(f" DNS Server: {dns}")
print(f" Lease Time: {lease_time} seconds")
print(" Interface ax0 configured successfully! 🚀")
print("----------------------------------------")
configured = True
lease_expiration = time.time() + int(lease_time)
def main():
global configured, lease_expiration
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.bind((LISTEN_IP, PORT))
print(f"Client {CLIENT_CALLSIGN} listening for CRAPRNIAC beacons...")
while True:
# Check lease expiration
if configured and lease_expiration and time.time() >= lease_expiration:
print("\n[Lease Expired]")
print(" Tearing down interface ax0...")
print(" Ready to rejoin network.")
print("----------------------------------------")
configured = False
lease_expiration = None
data, addr = sock.recvfrom(1024)
decoded = data.decode('utf-8')
parts = parse_message(decoded)
if len(parts) < 2:
continue
if parts[1] == "CRAP_BEACON":
if not configured:
base_callsign = parts[2]
network_name = parts[3]
print(f"\nHeard beacon from {base_callsign} ({addr}):")
print(f" Network: {network_name}")
print("Sending join request...")
request = build_request(network_name).encode('utf-8')
sock.sendto(request, addr)
elif parts[1] == "CRAP_ACCEPT":
client_callsign = parts[2]
network_name = parts[3]
assigned_ip = parts[4]
gateway = parts[5]
dns = parts[6]
lease_time = parts[7]
print(f"\nReceived CRAP_ACCEPT:")
print(f" Assigned IP: {assigned_ip}")
print(f" Gateway: {gateway}")
print(f" DNS Server: {dns}")
print(f" Lease Time: {lease_time} seconds")
apply_network_config(assigned_ip, gateway, dns, lease_time)
if __name__ == "__main__":
main()

29
pyproject.toml Normal file
View File

@ -0,0 +1,29 @@
[build-system]
requires = ["setuptools>=61.0"]
build-backend = "setuptools.build_meta"
[project]
name = "craprniac"
version = "0.1.0"
description = "Completely Ridiculous Amateur Protocol for Radio Node IP Auto-Configuration"
authors = [
{ name="John Burwell", email="your.email@example.com" }
]
license = { text = "MIT" }
readme = "README.md"
requires-python = ">=3.10"
dependencies = [
# put runtime dependencies here if needed later
]
[project.optional-dependencies]
dev = [
"pytest",
]
[tool.pytest.ini_options]
minversion = "6.0"
addopts = "-ra -q"
testpaths = [
"tests",
]