Key Code Examples
1. TCP Client-Server (Python)
import socket
import threading
def start_server(host='0.0.0.0', port=8080):
"""Multi-threaded TCP echo server."""
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server.bind((host, port))
server.listen(5)
print(f"Server listening on {host}:{port}")
while True:
client, addr = server.accept()
print(f"Connection from {addr}")
thread = threading.Thread(
target=handle_client, args=(client, addr)
)
thread.daemon = True
thread.start()
def handle_client(client, addr):
"""Handle individual client connection."""
try:
while True:
data = client.recv(4096)
if not data:
break
message = data.decode('utf-8')
print(f"Received from {addr}: {message}")
# Echo back with timestamp
response = f"[ECHO] {message}"
client.sendall(response.encode('utf-8'))
except ConnectionResetError:
print(f"Connection reset by {addr}")
finally:
client.close()
print(f"Connection closed: {addr}")
def start_client(server_host='127.0.0.1', server_port=8080):
"""Interactive TCP client."""
client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
client.connect((server_host, server_port))
print(f"Connected to {server_host}:{server_port}")
while True:
message = input("Message (or 'quit'): ")
if message.lower() == 'quit':
break
client.sendall(message.encode('utf-8'))
response = client.recv(4096)
print(f"Server: {response.decode('utf-8')}")
client.close()
# Run server in one terminal, client in another
# start_server()
# start_client()
2. UDP Server with Broadcasting
import socket
import struct
def udp_broadcast_server(port=9999):
"""UDP server that broadcasts messages to local network."""
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
sock.bind(('', port))
print(f"UDP broadcast server on port {port}")
while True:
data, addr = sock.recvfrom(1024)
message = data.decode('utf-8')
print(f"From {addr}: {message}")
# Broadcast to all on local network
broadcast_addr = ('255.255.255.255', port)
response = f"ACK: {message}".encode('utf-8')
sock.sendto(response, broadcast_addr)
def udp_client(server_ip='127.0.0.1', server_port=9999):
"""UDP client for sending messages."""
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.settimeout(5.0)
while True:
message = input("Send (or 'quit'): ")
if message.lower() == 'quit':
break
sock.sendto(message.encode('utf-8'),
(server_ip, server_port))
try:
data, addr = sock.recvfrom(1024)
print(f"Response: {data.decode('utf-8')}")
except socket.timeout:
print("No response received (timeout)")
sock.close()
3. HTTP Server from Scratch
import socket
import json
from urllib.parse import urlparse, parse_qs
class SimpleHTTPServer:
"""Minimal HTTP server implementing GET and POST."""
def __init__(self, host='0.0.0.0', port=8000):
self.host = host
self.port = port
self.routes = {}
def route(self, path, methods=['GET']):
"""Decorator to register route handlers."""
def decorator(func):
for method in methods:
self.routes[(path, method)] = func
return func
return decorator
def start(self):
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server.bind((self.host, self.port))
server.listen(5)
print(f"HTTP server on http://{self.host}:{self.port}")
while True:
client, addr = server.accept()
self.handle_request(client)
def handle_request(self, client):
try:
data = client.recv(8192).decode('utf-8')
lines = data.split('\r\n')
request_line = lines[0]
method, path, _ = request_line.split(' ')
# Parse URL and query parameters
parsed = urlparse(path)
params = parse_qs(parsed.query)
# Read body for POST
body = {}
if method == 'POST':
content_length = 0
for line in lines:
if line.lower().startswith('content-length:'):
content_length = int(line.split(':')[1].strip())
break
if content_length > 0:
body = json.loads(lines[-1])
# Find handler
handler = self.routes.get((parsed.path, method))
if handler:
status, response = handler(params, body)
else:
status, response = 404, {'error': 'Not Found'}
# Send response
self.send_response(client, status, response)
finally:
client.close()
def send_response(self, client, status, body):
status_text = {200: 'OK', 404: 'Not Found',
500: 'Internal Server Error'}.get(status, 'Unknown')
body_bytes = json.dumps(body).encode('utf-8')
response = (
f"HTTP/1.1 {status} {status_text}\r\n"
f"Content-Type: application/json\r\n"
f"Content-Length: {len(body_bytes)}\r\n"
f"Connection: close\r\n\r\n"
).encode('utf-8') + body_bytes
client.sendall(response)
# Usage
app = SimpleHTTPServer(port=8000)
@app.route('/', methods=['GET'])
def home(params, body):
return 200, {'message': 'Welcome', 'endpoints': ['/hello', '/add']}
@app.route('/hello', methods=['GET'])
def hello(params, body):
name = params.get('name', ['World'])[0]
return 200, {'greeting': f'Hello, {name}!'}
@app.route('/add', methods=['POST'])
def add(params, body):
a = body.get('a', 0)
b = body.get('b', 0)
return 200, {'result': a + b}
# app.start()
4. Subnet Calculator
import ipaddress
class SubnetCalculator:
"""Calculate network properties from CIDR notation."""
def __init__(self, cidr):
self.network = ipaddress.ip_network(cidr, strict=False)
def get_info(self):
return {
'network_address': str(self.network.network_address),
'broadcast_address': str(self.network.broadcast_address),
'subnet_mask': str(self.network.netmask),
'wildcard_mask': str(self.network.hostmask),
'num_hosts': self.network.num_addresses - 2,
'num_addresses': self.network.num_addresses,
'prefix_length': self.network.prefixlen,
'is_private': self.network.is_private,
'first_host': str(list(self.network.hosts())[0]),
'last_host': str(list(self.network.hosts())[-1]),
}
def get_subnets(self, new_prefix):
"""Divide network into smaller subnets."""
return [str(subnet) for subnet in self.network.subnets(
new_prefix=new_prefix
)]
def contains(self, ip):
"""Check if IP is in this network."""
return ipaddress.ip_address(ip) in self.network
# Examples
calc = SubnetCalculator('192.168.1.0/24')
info = calc.get_info()
for key, value in info.items():
print(f" {key}: {value}")
# Network: 192.168.1.0
# Broadcast: 192.168.1.255
# Subnet Mask: 255.255.255.0
# Hosts: 254
# Range: 192.168.1.1 - 192.168.1.254
# Create 4 subnets from /24
subnets = calc.get_subnets(26)
print(f"\nFour /26 subnets:")
for s in subnets:
print(f" {s}") # 192.168.1.0/26, .64/26, .128/26, .192/26
# Check IP membership
print(f"\n192.168.1.100 in network: {calc.contains('192.168.1.100')}") # True
print(f"10.0.0.1 in network: {calc.contains('10.0.0.1')}") # False
5. DNS Resolver
import socket
import struct
class DNSResolver:
"""Simple DNS resolver using raw UDP sockets."""
DNS_SERVER = "8.8.8.8" # Google DNS
DNS_PORT = 53
def build_query(self, domain, record_type=1):
"""Build DNS query packet."""
# Header: ID(2) + Flags(2) + QDCOUNT(2) + ANCOUNT(2)
# + NSCOUNT(2) + ARCOUNT(2) = 12 bytes
header = struct.pack('>HHHHHH',
0x1234, # Transaction ID
0x0100, # Flags: standard query
1, # Questions
0, # Answer RRs
0, # Authority RRs
0 # Additional RRs
)
# Question section
question = b''
for part in domain.split('.'):
question += bytes([len(part)]) + part.encode()
question += b'\x00' # End of name
# QTYPE (1=A, 5=CNAME, 28=AAAA) + QCLASS (1=IN)
question += struct.pack('>HH', record_type, 1)
return header + question
def resolve(self, domain):
"""Resolve domain name to IP address."""
query = self.build_query(domain)
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.settimeout(5)
sock.sendto(query, (self.DNS_SERVER, self.DNS_PORT))
response, _ = sock.recvfrom(512)
sock.close()
# Parse response (simplified)
# Skip header (12 bytes) + question section
# Count answers from header
ancount = struct.unpack('>H', response[6:8])[0]
return f"Response has {ancount} answers (raw parsing needed for IP)"
# Using standard library (simpler)
def resolve_domain(domain):
"""Resolve domain using system resolver."""
try:
ip = socket.gethostbyname(domain)
return f"{domain} -> {ip}"
except socket.gaierror as e:
return f"Failed to resolve {domain}: {e}"
# Examples
print(resolve_domain("www.google.com"))
print(resolve_domain("www.github.com"))
print(resolve_domain("localhost")) # 127.0.0.1
# Get all addresses
infos = socket.getaddrinfo("www.google.com", 80)
for info in infos[:3]:
print(f" Family: {info[0]}, IP: {info[4][0]}")
6. Port Scanner
import socket
import concurrent.futures
from datetime import datetime
class PortScanner:
"""Multi-threaded TCP port scanner."""
COMMON_PORTS = [21, 22, 23, 25, 53, 80, 110, 143, 443,
445, 3306, 3389, 5432, 6379, 8080, 8443, 27017]
def __init__(self, target, timeout=1.0, max_threads=100):
self.target = target
self.timeout = timeout
self.max_threads = max_threads
def scan_port(self, port):
"""Scan a single port."""
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(self.timeout)
try:
result = sock.connect_ex((self.target, port))
if result == 0:
service = self._identify_service(sock, port)
return port, True, service
return port, False, None
except (socket.timeout, OSError):
return port, False, None
finally:
sock.close()
def _identify_service(self, sock, port):
"""Try to identify service from banner."""
try:
sock.settimeout(2)
banner = sock.recv(1024).decode('utf-8', errors='ignore')
return banner.strip()[:50]
except:
return socket.getservbyport(port, 'tcp') if port < 1024 else "unknown"
def scan_range(self, start_port, end_port):
"""Scan a range of ports."""
print(f"Scanning {self.target}:{start_port}-{end_port}")
start_time = datetime.now()
open_ports = []
ports = range(start_port, end_port + 1)
with concurrent.futures.ThreadPoolExecutor(
max_workers=self.max_threads
) as executor:
futures = {executor.submit(self.scan_port, p): p for p in ports}
for future in concurrent.futures.as_completed(futures):
port, is_open, service = future.result()
if is_open:
print(f" [OPEN] Port {port}: {service}")
open_ports.append((port, service))
elapsed = (datetime.now() - start_time).total_seconds()
print(f"\nScan complete: {len(open_ports)} open ports in {elapsed:.2f}s")
return open_ports
def scan_common(self):
"""Scan common ports."""
return self.scan_range(1, 1024)
# Usage (for educational purposes only — scan your own systems)
# scanner = PortScanner("127.0.0.1")
# scanner.scan_common()
7. Packet Sniffer (Raw Sockets)
import socket
import struct
def parse_ip_header(data):
"""Parse IPv4 header fields."""
version_ihl = data[0]
version = version_ihl >> 4
ihl = (version_ihl & 0xF) * 4
total_length = struct.unpack('!H', data[2:4])[0]
protocol = data[9]
src_ip = socket.inet_ntoa(data[12:16])
dst_ip = socket.inet_ntoa(data[16:20])
return {
'version': version,
'header_length': ihl,
'total_length': total_length,
'protocol': protocol,
'src_ip': src_ip,
'dst_ip': dst_ip,
'payload': data[ihl:]
}
def parse_tcp_header(data):
"""Parse TCP header fields."""
if len(data) < 20:
return None
src_port = struct.unpack('!H', data[0:2])[0]
dst_port = struct.unpack('!H', data[2:4])[0]
seq_num = struct.unpack('!I', data[4:8])[0]
ack_num = struct.unpack('!I', data[8:12])[0]
data_offset = (data[12] >> 4) * 4
flags = data[13]
flag_str = ''
if flags & 0x02: flag_str += 'SYN '
if flags & 0x10: flag_str += 'ACK '
if flags & 0x01: flag_str += 'FIN '
if flags & 0x08: flag_str += 'PSH '
if flags & 0x04: flag_str += 'RST '
return {
'src_port': src_port,
'dst_port': dst_port,
'seq': seq_num,
'ack': ack_num,
'flags': flag_str.strip(),
'payload': data[data_offset:]
}
def packet_sniffer(interface=None, count=10):
"""Capture and parse packets (requires admin/root)."""
# Create raw socket
sock = socket.socket(socket.AF_INET, socket.SOCK_RAW,
socket.IPPROTO_TCP)
if interface:
sock.bind((interface, 0))
sock.settimeout(10)
print(f"Capturing {count} packets...")
for i in range(count):
try:
data, addr = sock.recvfrom(65535)
ip_info = parse_ip_header(data)
proto_name = {6: 'TCP', 17: 'UDP', 1: 'ICMP'}.get(
ip_info['protocol'], f"Other({ip_info['protocol']})"
)
print(f"\nPacket {i+1}: {ip_info['src_ip']} -> {ip_info['dst_ip']}")
print(f" Protocol: {proto_name}, Length: {ip_info['total_length']}")
if ip_info['protocol'] == 6: # TCP
tcp_info = parse_tcp_header(ip_info['payload'])
if tcp_info:
print(f" TCP: {tcp_info['src_port']} -> {tcp_info['dst_port']}")
print(f" Flags: {tcp_info['flags']}")
except socket.timeout:
break
sock.close()
# packet_sniffer() # Requires root/admin privileges
8. WebSocket Server (Real-time Communication)
import asyncio
import websockets
import json
from datetime import datetime
class WebSocketServer:
"""Real-time WebSocket server for chat/collaboration."""
def __init__(self):
self.clients = {} # websocket -> user_info
self.rooms = {} # room_name -> set of websockets
async def register(self, websocket, username):
self.clients[websocket] = {
'username': username,
'joined_at': datetime.now().isoformat()
}
await self.broadcast_system(f"{username} joined the chat")
async def unregister(self, websocket):
if websocket in self.clients:
username = self.clients[websocket]['username']
del self.clients[websocket]
# Remove from rooms
for room in self.rooms.values():
room.discard(websocket)
await self.broadcast_system(f"{username} left the chat")
async def broadcast(self, message, sender=None):
"""Send message to all connected clients."""
for websocket in self.clients:
if websocket != sender:
await websocket.send(json.dumps(message))
async def broadcast_system(self, content):
await self.broadcast({
'type': 'system',
'content': content,
'timestamp': datetime.now().isoformat(),
'online_count': len(self.clients)
})
async def handle_message(self, websocket, data):
"""Process incoming message."""
user = self.clients[websocket]['username']
if data['type'] == 'chat':
await self.broadcast({
'type': 'chat',
'from': user,
'content': data['content'],
'timestamp': datetime.now().isoformat()
}, sender=websocket)
elif data['type'] == 'join_room':
room = data['room']
if room not in self.rooms:
self.rooms[room] = set()
self.rooms[room].add(websocket)
await self.broadcast_system(f"{user} joined #{room}")
async def handler(self, websocket, path):
"""Main WebSocket connection handler."""
try:
# Wait for registration
reg_msg = await websocket.recv()
reg_data = json.loads(reg_msg)
await self.register(websocket, reg_data['username'])
# Message loop
async for message in websocket:
data = json.loads(message)
await self.handle_message(websocket, data)
except websockets.exceptions.ConnectionClosed:
pass
finally:
await self.unregister(websocket)
def start(self, host='0.0.0.0', port=8765):
print(f"WebSocket server on ws://{host}:{port}")
start_server = websockets.serve(self.handler, host, port)
asyncio.get_event_loop().run_until_complete(start_server)
asyncio.get_event_loop().run_forever()
# Usage
# server = WebSocketServer()
# server.start()
9. Network Configuration (Netmiko/SSH)
from netmiko import ConnectHandler
import json
class NetworkAdmin:
"""SSH-based network device administration."""
def __init__(self, device_type, host, username, password):
self.device = {
'device_type': device_type,
'host': host,
'username': username,
'password': password,
}
def connect(self):
self.connection = ConnectHandler(**self.device)
print(f"Connected to {self.device['host']}")
def get_interfaces(self):
"""Get interface status."""
output = self.connection.send_command(
'show ip int brief', use_textfsm=True
)
return output
def get_routing_table(self):
"""Get routing table."""
output = self.connection.send_command(
'show ip route', use_textfsm=True
)
return output
def configure_vlan(self, vlan_id, name):
"""Create a VLAN."""
commands = [
f'vlan {vlan_id}',
f'name {name}',
'exit'
]
output = self.connection.send_config_set(commands)
return output
def configure_interface(self, interface, ip_address, subnet_mask):
"""Configure IP on interface."""
commands = [
f'interface {interface}',
f'ip address {ip_address} {subnet_mask}',
'no shutdown',
'exit'
]
return self.connection.send_config_set(commands)
def backup_config(self):
"""Backup running configuration."""
config = self.connection.send_command('show running-config')
filename = f"backup_{self.device['host']}_{datetime.now():%Y%m%d}.cfg"
with open(filename, 'w') as f:
f.write(config)
print(f"Backup saved: {filename}")
return config
def close(self):
self.connection.disconnect()
# Usage
# admin = NetworkAdmin('cisco_ios', '192.168.1.1', 'admin', 'password')
# admin.connect()
# interfaces = admin.get_interfaces()
# admin.configure_vlan(10, 'Engineering')
# admin.backup_config()
# admin.close()
10. Load Balancer Implementation
import socket
import threading
import hashlib
import itertools
class LoadBalancer:
"""TCP load balancer with multiple strategies."""
def __init__(self, listen_port, strategy='round_robin'):
self.port = listen_port
self.strategy = strategy
self.backends = []
self.healthy_backends = []
self.rr_counter = itertools.count()
self.session_table = {} # client_ip -> backend
self.lock = threading.Lock()
def add_backend(self, host, port, weight=1):
"""Add a backend server."""
self.backends.append({
'host': host,
'port': port,
'weight': weight,
'healthy': True,
'connections': 0
})
self.healthy_backends = [b for b in self.backends if b['healthy']]
def _round_robin(self):
"""Select backend using round-robin."""
if not self.healthy_backends:
return None
idx = next(self.rr_counter) % len(self.healthy_backends)
return self.healthy_backends[idx]
def _least_connections(self):
"""Select backend with fewest active connections."""
if not self.healthy_backends:
return None
return min(self.healthy_backends, key=lambda b: b['connections'])
def _ip_hash(self, client_ip):
"""Select backend based on client IP hash."""
if not self.healthy_backends:
return None
hash_val = int(hashlib.md5(client_ip.encode()).hexdigest(), 16)
return self.healthy_backends[hash_val % len(self.healthy_backends)]
def select_backend(self, client_ip):
"""Select backend based on strategy."""
if self.strategy == 'round_robin':
return self._round_robin()
elif self.strategy == 'least_connections':
return self._least_connections()
elif self.strategy == 'ip_hash':
return self._ip_hash(client_ip)
return self._round_robin()
def health_check(self):
"""Check health of all backends."""
for backend in self.backends:
try:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(2)
sock.connect((backend['host'], backend['port']))
sock.close()
backend['healthy'] = True
except (socket.error, OSError):
backend['healthy'] = False
self.healthy_backends = [b for b in self.backends if b['healthy']]
def handle_client(self, client_sock, client_addr):
"""Proxy client connection to selected backend."""
backend = self.select_backend(client_addr[0])
if not backend:
client_sock.sendall(b"HTTP/1.1 503 Service Unavailable\r\n\r\n")
client_sock.close()
return
try:
# Connect to backend
backend_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
backend_sock.connect((backend['host'], backend['port']))
backend['connections'] += 1
# Bidirectional proxy
def forward(src, dst):
try:
while True:
data = src.recv(8192)
if not data:
break
dst.sendall(data)
except OSError:
pass
finally:
src.close()
dst.close()
t1 = threading.Thread(target=forward,
args=(client_sock, backend_sock))
t2 = threading.Thread(target=forward,
args=(backend_sock, client_sock))
t1.start()
t2.start()
t1.join()
t2.join()
finally:
backend['connections'] -= 1
def start(self):
"""Start the load balancer."""
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server.bind(('0.0.0.0', self.port))
server.listen(100)
print(f"Load balancer on :{self.port} ({self.strategy})")
print(f"Backends: {len(self.backends)}")
while True:
client, addr = server.accept()
thread = threading.Thread(
target=self.handle_client, args=(client, addr)
)
thread.daemon = True
thread.start()
# Usage
# lb = LoadBalancer(8080, strategy='round_robin')
# lb.add_backend('10.0.1.10', 3000)
# lb.add_backend('10.0.1.11', 3000)
# lb.add_backend('10.0.1.12', 3000)
# lb.start()
Core Concepts
OSI Model (7 Layers)
Definition: The OSI model is a conceptual framework that standardizes network communication into 7 abstraction layers.
- Layer 7 — Application: HTTP, FTP, SMTP, DNS — user-facing protocols
- Layer 6 — Presentation: Data formatting, encryption (SSL/TLS), compression
- Layer 5 — Session: Session establishment, management, termination
- Layer 4 — Transport: TCP (reliable, connection-oriented), UDP (fast, connectionless)
- Layer 3 — Network: IP addressing, routing, logical addressing (routers)
- Layer 2 — Data Link: MAC addresses, Ethernet frames, switches, error detection
- Layer 1 — Physical: Cables, signals, bits, electrical/optical transmission
Mnemonic: "Please Do Not Throw Sausage Pizza Away" (bottom to top)
TCP vs UDP
| Feature | TCP | UDP |
|---|---|---|
| Connection | Connection-oriented | Connectionless |
| Reliability | Guaranteed delivery | Best-effort |
| Ordering | In-order delivery | No ordering |
| Speed | Slower (overhead) | Faster (minimal overhead) |
| Use Cases | Web, email, file transfer | Video streaming, gaming, DNS |
| Header Size | 20-60 bytes | 8 bytes |
TCP Three-Way Handshake: SYN → SYN-ACK → ACK
IP Addressing & Subnetting
IPv4 Address: 32-bit address, written as 4 octets (e.g., 192.168.1.1)
- Class A: 1.0.0.0 — 126.255.255.255 (/8), 16M hosts
- Class B: 128.0.0.0 — 191.255.255.255 (/16), 65K hosts
- Class C: 192.0.0.0 — 223.255.255.255 (/24), 254 hosts
- Private Ranges: 10.0.0.0/8, 172.16.0.0/12, 192.168.0.0/16
IPv6: 128-bit address (e.g., 2001:0db8:85a3::8a2e:0370:7334). Solves IPv4 exhaustion.
CIDR Notation: /24 means first 24 bits are network, leaving 8 bits for hosts (256 addresses).
Routing Protocols
- RIP (Routing Information Protocol): Distance-vector, hop count metric, max 15 hops
- OSPF (Open Shortest Path First): Link-state, Dijkstra's algorithm, fast convergence
- BGP (Border Gateway Protocol): Path-vector, used for inter-AS routing (the internet's routing protocol)
- EIGRP (Enhanced Interior Gateway): Cisco proprietary, hybrid protocol
DNS (Domain Name System)
Definition: Hierarchical, distributed naming system that translates domain names to IP addresses.
- Root Servers: 13 logical root servers (A-M) worldwide
- TLD Servers: Top-Level Domain servers (.com, .org, .edu, country codes)
- Authoritative Servers: Hold actual DNS records for domains
- Recursive Resolver: Client's ISP resolver that queries on behalf of client
- DNS Records: A (IPv4), AAAA (IPv6), CNAME (alias), MX (mail), TXT (text), NS (nameserver)
- Resolution Process: Browser cache → OS cache → Resolver → Root → TLD → Authoritative
Network Security Fundamentals
- Firewall: Filters traffic based on rules (packet filter, stateful, application-layer)
- VPN (Virtual Private Network): Encrypted tunnel over public network (IPSec, SSL/TLS, WireGuard)
- IDS/IPS: Intrusion Detection/Prevention Systems
- DMZ: Demilitarized Zone — network segment between internal and external
- DDoS Protection: Mitigates distributed denial-of-service attacks
- 802.1X: Network access control for port-based authentication
Wireless Networking (Wi-Fi)
- 802.11a: 5 GHz, 54 Mbps
- 802.11b/g: 2.4 GHz, 11/54 Mbps
- 802.11n (Wi-Fi 4): 2.4/5 GHz, MIMO, 600 Mbps
- 802.11ac (Wi-Fi 5): 5 GHz, 6.93 Gbps
- 802.11ax (Wi-Fi 6/6E): 2.4/5/6 GHz, 9.6 Gbps, OFDMA
- Security: WPA3 (latest), WPA2, WEP (broken)
- Concepts: SSID, channels, bandwidth, MIMO, beamforming
Network Topology Types
- Star: All devices connect to central switch/hub — most common in LANs
- Bus: All devices on single cable — outdated, collisions
- Ring: Devices in circular chain — token passing (Token Ring, FDDI)
- Mesh: Every device connects to every other — maximum redundancy
- Tree: Hierarchical star topologies — scalable
- Hybrid: Combination of two or more topologies
Worked Examples
Example 1: Subnet Calculation
Problem: You need to create 6 subnets from 192.168.0.0/24, each with at least 30 hosts. Determine the new subnet mask and ranges.
Solution:
- Need 6 subnets → borrow 3 bits (2³ = 8 ≥ 6)
- New prefix: /24 + 3 = /27
- Subnet mask: 255.255.255.224
- Hosts per subnet: 2⁵ - 2 = 30 hosts
- 8 Subnets:
- 192.168.1.0/27 → hosts .1-.30, broadcast .31
- 192.168.1.32/27 → hosts .33-.62, broadcast .63
- 192.168.1.64/27 → hosts .65-.94, broadcast .95
- ... (continuing in increments of 32)
Example 2: TCP Handshake Analysis
Problem: Explain what happens when a browser connects to https://www.example.com
Solution:
- 1. DNS Lookup: Browser resolves www.example.com → 93.184.216.34
- 2. TCP Handshake:
- Client → Server: SYN (seq=100)
- Server → Client: SYN-ACK (seq=200, ack=101)
- Client → Server: ACK (seq=101, ack=201)
- 3. TLS Handshake: ClientHello → ServerHello → Certificate → Key Exchange → Finished
- 4. HTTP Request: GET / HTTP/1.1 (encrypted in TLS)
- 5. HTTP Response: 200 OK with HTML content
Example 3: Calculate Network Bandwidth
Problem: A 1 Gbps link carries 1500-byte packets. How many packets per second can it handle?
Solution:
- Link capacity: 1 Gbps = 1,000,000,000 bits/second
- Packet size: 1500 bytes × 8 = 12,000 bits
- Packets per second: 1,000,000,000 / 12,000 = 83,333 pps
- Answer: ~83,333 packets per second
- With 64-byte minimum packets: 1,000,000,000 / 512 = 1,953,125 pps
Example 4: Design a Corporate Network
Problem: Design a network for a 3-floor office with 50 users per floor.
Solution:
- Core Layer: 2× core switches (redundant), 10 Gbps backbone
- Distribution Layer: 1 switch per floor, 1 Gbps to access
- Access Layer: 2× 48-port switches per floor (PoE for VoIP)
- VLANs: VLAN 10 (Eng), VLAN 20 (Sales), VLAN 30 (Guest), VLAN 99 (Management)
- Subnets: 10.10.10.0/24, 10.10.20.0/24, 10.10.30.0/24
- Routing: OSPF between distribution and core
- Security: 802.1X port auth, firewall at perimeter, IPS/IDS
Example 5: NAT Configuration
Problem: Configure NAT for a small office with public IP 203.0.113.10 and internal network 192.168.1.0/24.
! Cisco NAT Configuration
! Define inside and outside interfaces
interface GigabitEthernet0/0
ip address 192.168.1.1 255.255.255.0
ip nat inside
interface GigabitEthernet0/1
ip address 203.0.113.10 255.255.255.0
ip nat outside
! Dynamic NAT with overload (PAT)
access-list 1 permit 192.168.1.0 0.0.0.255
ip nat inside source list 1 interface GigabitEthernet0/1 overload
! Static NAT for web server (192.168.1.100)
ip nat inside source static tcp 192.168.1.100 80 \
203.0.113.10 80 extendable
ip nat inside source static tcp 192.168.1.100 443 \
203.0.113.10 443 extendable
Practice Problems
Problem 1: Calculate Subnet Parameters
For network 172.16.32.0/20, find: network address, broadcast address, number of hosts, subnet mask, and first/last usable IPs.
Answer:
- Network: 172.16.32.0
- Broadcast: 172.16.47.255
- Subnet Mask: 255.255.240.0
- Hosts: 2¹² - 2 = 4,094
- First usable: 172.16.32.1
- Last usable: 172.16.47.254
Problem 2: Implement HTTP Keep-Alive
Write a Python TCP server that supports HTTP/1.1 keep-alive connections.
import socket
import threading
def handle_keepalive(client):
client.settimeout(30)
while True:
try:
data = client.recv(4096)
if not data:
break
response = (
"HTTP/1.1 200 OK\r\n"
"Content-Type: text/html\r\n"
"Connection: keep-alive\r\n"
"Content-Length: 13\r\n"
"\r\n"
"Hello, World!"
)
client.sendall(response.encode())
except socket.timeout:
break
client.close()
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server.bind(('0.0.0.0', 8080))
server.listen(5)
while True:
client, addr = server.accept()
threading.Thread(target=handle_keepalive,
args=(client,)).start()
Problem 3: VLAN Configuration
Configure VLANs 10 (Engineering), 20 (Sales), and 30 (Guest) on a Cisco switch with ports 1-8 for Eng, 9-16 for Sales, 17-24 for Guest.
vlan 10
name Engineering
vlan 20
name Sales
vlan 30
name Guest
interface range fa0/1-8
switchport mode access
switchport access vlan 10
spanning-tree portfast
interface range fa0/9-16
switchport mode access
switchport access vlan 20
spanning-tree portfast
interface range fa0/17-24
switchport mode access
switchport access vlan 30
spanning-tree portfast
Problem 4: Troubleshoot Connectivity
A user cannot reach www.google.com (8.8.8.8) but can ping their default gateway 192.168.1.1. What are possible causes and diagnostic steps?
Answer:
- Possible Causes: DNS failure, ISP outage, firewall blocking, routing issue, MTU problems
- Diagnostics:
- ping 8.8.8.8 — tests basic connectivity
- nslookup google.com — tests DNS resolution
- tracert 8.8.8.8 — shows where packets stop
- Check firewall rules
- Test with different browser/device
Problem 5: Calculate CIDR for Requirements
What CIDR prefix is needed for each requirement: (a) 500 hosts, (b) 1000 hosts, (c) 10 subnets with 100 hosts each?
Answer:
- (a) 500 hosts → 2⁹-2=510 ≥ 500 → /23 (512 addresses)
- (b) 1000 hosts → 2¹⁰-2=1022 ≥ 1000 → /22 (1024 addresses)
- (c) 10 subnets with 100 hosts each → Need /25 per subnet (126 hosts), 10 subnets = 2⁴=16 ≥ 10 → /21 total
Problem 6: Implement a Simple Proxy Server
Create a TCP proxy server in Python that forwards traffic between a client and a target server.
import socket
import threading
def proxy_handler(client_sock, target_host, target_port):
target_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
target_sock.connect((target_host, target_port))
def forward(src, dst):
try:
while True:
data = src.recv(8192)
if not data:
break
dst.sendall(data)
except OSError:
pass
threading.Thread(target=forward, args=(client_sock, target_sock)).start()
threading.Thread(target=forward, args=(target_sock, client_sock)).start()
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server.bind(('0.0.0.0', 8888))
server.listen(50)
print("Proxy on :8888")
while True:
client, addr = server.accept()
threading.Thread(target=proxy_handler,
args=(client, 'example.com', 80)).start()
Problem 7: QoS Configuration
Configure QoS on a router to prioritize VoIP traffic (UDP 5060, RTP 10000-20000) over regular web traffic.
! Define class maps
class-map match-any VOIP
match protocol sip
match protocol rtp audio
class-map match-any WEB
match protocol http
match protocol https
! Define policy map
policy-map QOS_POLICY
class VOIP
priority percent 30
set dscp ef
class WEB
bandwidth percent 40
set dscp af21
class class-default
bandwidth percent 25
! Apply to interface
interface GigabitEthernet0/1
service-policy output QOS_POLICY
Problem 8: Design a Firewall Ruleset
Create firewall rules for: allow HTTP/HTTPS outbound, allow SSH from management subnet only, deny all other inbound.
# iptables rules
iptables -P INPUT DROP
iptables -P FORWARD DROP
iptables -P OUTPUT ACCEPT
# Allow established connections
iptables -A INPUT -m state --state ESTABLISHED,RELATED -j ACCEPT
# Allow loopback
iptables -A INPUT -i lo -j ACCEPT
# Allow SSH from management subnet only
iptables -A INPUT -p tcp --dport 22 -s 10.0.0.0/24 -j ACCEPT
# Allow HTTP/HTTPS inbound (web server)
iptables -A INPUT -p tcp --dport 80 -j ACCEPT
iptables -A INPUT -p tcp --dport 443 -j ACCEPT
# Rate limiting (prevent brute force)
iptables -A INPUT -p tcp --dport 22 -m recent --set --name ssh
iptables -A INPUT -p tcp --dport 22 -m recent --update --seconds 60 \
--hitcount 4 --name ssh -j DROP
Problem 9: Wi-Fi Channel Planning
Plan Wi-Fi channels for a 4-access-point deployment in the 2.4 GHz band (channels 1, 6, 11 available).
Answer:
- 2.4 GHz: Only 3 non-overlapping channels (1, 6, 11)
- Assign: AP1=Ch1, AP2=Ch6, AP3=Ch11, AP4=Ch1 (far from AP1)
- Minimize co-channel interference by placing same-channel APs far apart
- Use 5 GHz for high-density areas (more non-overlapping channels)
Problem 10: Design a CDN Strategy
Design a CDN strategy for a global e-commerce site with users in North America, Europe, and Asia.
Answer:
- Edge Locations: Deploy to 50+ PoPs worldwide
- Static Assets: Cache images, CSS, JS with 1-year TTL + versioning
- Dynamic Content: Use stale-while-revalidate for product pages
- Origin Shield: Additional caching layer between edge and origin
- Load Balancing: GeoDNS routes users to nearest PoP