Prerequisites
- Basic understanding of programming concepts 📝
- Python installation (3.8+) 🐍
- VS Code or preferred IDE 💻
What you'll learn
- Understand port scanning fundamentals 🎯
- Apply network discovery in real projects 🏗️
- Debug common networking issues 🐛
- Write clean, Pythonic network code ✨
🎯 Introduction
Welcome to this exciting tutorial on port scanning and network discovery! 🎉 In this guide, we’ll explore how to discover services and open ports on network hosts using Python.
You’ll discover how port scanning can transform your network administration and security testing capabilities. Whether you’re monitoring network infrastructure 🖥️, performing security audits 🔒, or troubleshooting connectivity issues 🔧, understanding port scanning is essential for network professionals.
By the end of this tutorial, you’ll feel confident implementing network discovery tools in your own projects! Let’s dive in! 🏊♂️
📚 Understanding Port Scanning
🤔 What is Port Scanning?
Port scanning is like checking which doors and windows are open in a building 🏢. Think of it as a network reconnaissance technique that helps you discover which services are available on a host.
In networking terms, port scanning involves systematically probing a server or host for open ports 🔍. This means you can:
- ✨ Discover running services
- 🚀 Map network topology
- 🛡️ Identify potential vulnerabilities
💡 Why Use Port Scanning?
Here’s why network professionals love port scanning:
- Network Discovery 🔍: Find active hosts and services
- Security Auditing 🔒: Identify exposed services
- Troubleshooting 🔧: Diagnose connectivity issues
- Service Monitoring 📊: Track service availability
Real-world example: Imagine managing a corporate network 🏢. With port scanning, you can quickly identify unauthorized services or verify that critical services are accessible.
🔧 Basic Syntax and Usage
📝 Simple Port Check
Let’s start with a friendly example:
# 👋 Hello, Network Discovery!
import socket
import sys
def check_port(host, port):
"""
Check if a port is open on a host
"""
# 🎨 Create a socket object
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(1) # ⏱️ 1 second timeout
try:
# 🔌 Attempt to connect
result = sock.connect_ex((host, port))
sock.close()
# ✅ Port is open!
if result == 0:
return True
else:
return False
except socket.gaierror:
# 🚫 Hostname couldn't be resolved
print(f"❌ Hostname {host} could not be resolved")
return False
except socket.error:
# 💥 Couldn't connect to server
print(f"❌ Could not connect to {host}")
return False
# 🎮 Let's test it!
if check_port("google.com", 80):
print("✅ Port 80 is open on google.com!")
else:
print("❌ Port 80 is closed on google.com")💡 Explanation: Notice how we use socket programming to attempt connections! The connect_ex() method returns 0 if the connection succeeds.
🎯 Common Port Scanning Patterns
Here are patterns you’ll use frequently:
# 🏗️ Pattern 1: Multiple port scanning
import concurrent.futures
import socket
def scan_port(host, port):
"""Scan a single port"""
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(0.5)
result = sock.connect_ex((host, port))
sock.close()
return port if result == 0 else None
def scan_ports(host, start_port, end_port, max_threads=100):
"""
# 🚀 Scan multiple ports concurrently
"""
open_ports = []
# 🎯 Use thread pool for concurrent scanning
with concurrent.futures.ThreadPoolExecutor(max_workers=max_threads) as executor:
# 📊 Submit all port scans
futures = {
executor.submit(scan_port, host, port): port
for port in range(start_port, end_port + 1)
}
# 🔄 Collect results
for future in concurrent.futures.as_completed(futures):
port = future.result()
if port:
open_ports.append(port)
print(f"✨ Found open port: {port}")
return sorted(open_ports)
# 🎨 Pattern 2: Service identification
def identify_service(port):
"""Identify common services by port number"""
services = {
21: "FTP 📁",
22: "SSH 🔐",
23: "Telnet 📟",
25: "SMTP 📧",
53: "DNS 🌐",
80: "HTTP 🌍",
110: "POP3 📮",
443: "HTTPS 🔒",
3306: "MySQL 🗄️",
3389: "RDP 🖥️",
8080: "HTTP-Proxy 🔄"
}
return services.get(port, "Unknown ❓")💡 Practical Examples
🛡️ Example 1: Network Service Scanner
Let’s build a comprehensive network scanner:
# 🛡️ Advanced network service scanner
import socket
import threading
from datetime import datetime
import queue
class NetworkScanner:
def __init__(self, host, port_range=(1, 1000), threads=50):
self.host = host
self.port_range = port_range
self.threads = threads
self.open_ports = []
self.lock = threading.Lock()
def scan_port(self, port):
"""
# 🔍 Scan a single port
"""
try:
# 🎯 Create socket with timeout
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(1)
# 🔌 Try to connect
result = sock.connect_ex((self.host, port))
sock.close()
if result == 0:
# ✅ Port is open!
service = self.get_service_banner(port)
with self.lock:
self.open_ports.append({
'port': port,
'service': identify_service(port),
'banner': service
})
print(f"✨ Port {port}: OPEN - {identify_service(port)}")
except Exception as e:
pass # 🤫 Silently skip errors
def get_service_banner(self, port):
"""
# 📋 Try to grab service banner
"""
try:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(2)
sock.connect((self.host, port))
# 👋 Send a probe
sock.send(b'Hello\r\n')
banner = sock.recv(1024).decode('utf-8', errors='ignore').strip()
sock.close()
return banner[:50] if banner else ""
except:
return ""
def worker(self, port_queue):
"""
# 🔄 Worker thread for scanning
"""
while True:
port = port_queue.get()
if port is None:
break
self.scan_port(port)
port_queue.task_done()
def scan(self):
"""
# 🚀 Start the scanning process
"""
print(f"🎯 Starting scan on {self.host}")
print(f"📊 Port range: {self.port_range[0]}-{self.port_range[1]}")
print(f"⚡ Using {self.threads} threads\n")
start_time = datetime.now()
# 📦 Create work queue
port_queue = queue.Queue()
# 🎨 Create and start threads
threads = []
for _ in range(self.threads):
t = threading.Thread(target=self.worker, args=(port_queue,))
t.start()
threads.append(t)
# 📝 Add ports to queue
for port in range(self.port_range[0], self.port_range[1] + 1):
port_queue.put(port)
# 🏁 Wait for completion
port_queue.join()
# 🛑 Stop workers
for _ in range(self.threads):
port_queue.put(None)
for t in threads:
t.join()
# 📊 Display results
duration = datetime.now() - start_time
self.display_results(duration)
def display_results(self, duration):
"""
# 📋 Display scan results
"""
print(f"\n{'='*50}")
print(f"🎉 Scan Complete!")
print(f"⏱️ Duration: {duration}")
print(f"🎯 Host: {self.host}")
print(f"✅ Open ports found: {len(self.open_ports)}")
print(f"{'='*50}\n")
if self.open_ports:
print("📊 Open Ports Summary:")
print(f"{'Port':<10} {'Service':<20} {'Banner':<30}")
print("-" * 60)
for port_info in sorted(self.open_ports, key=lambda x: x['port']):
print(f"{port_info['port']:<10} {port_info['service']:<20} {port_info['banner']:<30}")
else:
print("❌ No open ports found in the specified range")
# 🎮 Let's use it!
scanner = NetworkScanner("scanme.nmap.org", port_range=(1, 100))
scanner.scan()🎯 Try it yourself: Add UDP port scanning capability and OS fingerprinting features!
🌐 Example 2: Subnet Discovery Tool
Let’s scan an entire subnet:
# 🌐 Subnet discovery tool
import ipaddress
import concurrent.futures
import socket
import platform
import subprocess
class SubnetScanner:
def __init__(self, subnet):
self.subnet = ipaddress.ip_network(subnet, strict=False)
self.active_hosts = []
def ping_host(self, ip):
"""
# 🏓 Ping a host to check if it's alive
"""
ip_str = str(ip)
# 🖥️ Platform-specific ping command
if platform.system().lower() == "windows":
command = ["ping", "-n", "1", "-w", "1000", ip_str]
else:
command = ["ping", "-c", "1", "-W", "1", ip_str]
try:
# 🚀 Execute ping
result = subprocess.run(
command,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL
)
if result.returncode == 0:
return ip_str
except:
pass
return None
def tcp_scan_host(self, ip):
"""
# 🔌 TCP scan to check if host is alive
"""
common_ports = [80, 443, 22, 21, 25, 3389, 445, 139]
ip_str = str(ip)
for port in common_ports:
try:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(0.5)
result = sock.connect_ex((ip_str, port))
sock.close()
if result == 0:
return ip_str
except:
continue
return None
def scan_subnet(self, method="tcp"):
"""
# 🌍 Scan entire subnet for active hosts
"""
print(f"🎯 Scanning subnet: {self.subnet}")
print(f"📊 Total hosts to scan: {self.subnet.num_addresses - 2}")
print(f"⚡ Method: {method.upper()}\n")
# 🎨 Choose scanning method
scan_func = self.ping_host if method == "ping" else self.tcp_scan_host
# 🚀 Concurrent scanning
with concurrent.futures.ThreadPoolExecutor(max_workers=100) as executor:
# 📝 Submit all scans
future_to_ip = {
executor.submit(scan_func, ip): ip
for ip in self.subnet.hosts()
}
# 🔄 Process results
for future in concurrent.futures.as_completed(future_to_ip):
result = future.result()
if result:
self.active_hosts.append(result)
print(f"✅ Found active host: {result}")
# 📊 Display summary
self.display_summary()
def display_summary(self):
"""
# 📋 Display scanning summary
"""
print(f"\n{'='*50}")
print(f"🎉 Subnet Scan Complete!")
print(f"🌐 Subnet: {self.subnet}")
print(f"✅ Active hosts: {len(self.active_hosts)}")
print(f"❌ Inactive hosts: {self.subnet.num_addresses - 2 - len(self.active_hosts)}")
print(f"{'='*50}\n")
if self.active_hosts:
print("📊 Active Hosts:")
for host in sorted(self.active_hosts, key=ipaddress.ip_address):
print(f" 🖥️ {host}")
# 🎮 Let's discover a network!
scanner = SubnetScanner("192.168.1.0/24")
scanner.scan_subnet(method="tcp")🚀 Advanced Concepts
🧙♂️ SYN Stealth Scanning
When you’re ready to level up, try stealth scanning:
# 🎯 Advanced SYN stealth scanner (requires root/admin)
from scapy.all import *
import logging
# 🤫 Suppress scapy warnings
logging.getLogger("scapy.runtime").setLevel(logging.ERROR)
class StealthScanner:
def __init__(self, target, ports):
self.target = target
self.ports = ports
self.open_ports = []
def syn_scan(self, port):
"""
# 🥷 Perform SYN stealth scan
"""
# 🎨 Craft SYN packet
syn_packet = IP(dst=self.target) / TCP(dport=port, flags="S")
# 🚀 Send packet and wait for response
response = sr1(syn_packet, timeout=1, verbose=0)
if response and response.haslayer(TCP):
if response[TCP].flags == 18: # SYN-ACK
# ✅ Port is open!
self.open_ports.append(port)
# 🛑 Send RST to close connection
rst_packet = IP(dst=self.target) / TCP(dport=port, flags="R")
send(rst_packet, verbose=0)
return True
return False
def scan(self):
"""
# 🚀 Execute stealth scan
"""
print(f"🥷 Starting SYN stealth scan on {self.target}")
for port in self.ports:
if self.syn_scan(port):
print(f"✨ Port {port}: OPEN (Stealth)")
print(f"\n🎯 Stealth scan complete!")
print(f"✅ Open ports: {self.open_ports}")
# ⚠️ Note: Requires root/admin privileges!
# scanner = StealthScanner("scanme.nmap.org", [80, 443, 22, 21])
# scanner.scan()🏗️ Advanced Service Detection
For the brave network explorers:
# 🚀 Advanced service fingerprinting
import struct
import ssl
class ServiceDetector:
def __init__(self, host, port):
self.host = host
self.port = port
def detect_http(self):
"""
# 🌐 Detect HTTP/HTTPS service
"""
try:
# 🔒 Try HTTPS first
context = ssl.create_default_context()
with socket.create_connection((self.host, self.port), timeout=3) as sock:
with context.wrap_socket(sock, server_hostname=self.host) as ssock:
ssock.send(b"HEAD / HTTP/1.1\r\nHost: " + self.host.encode() + b"\r\n\r\n")
response = ssock.recv(1024).decode('utf-8', errors='ignore')
if "HTTP" in response:
return f"HTTPS 🔒 - {response.split('\r\n')[0]}"
except:
# 🌍 Try plain HTTP
try:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(3)
sock.connect((self.host, self.port))
sock.send(b"HEAD / HTTP/1.1\r\nHost: " + self.host.encode() + b"\r\n\r\n")
response = sock.recv(1024).decode('utf-8', errors='ignore')
sock.close()
if "HTTP" in response:
return f"HTTP 🌍 - {response.split('\r\n')[0]}"
except:
pass
return None
def detect_ssh(self):
"""
# 🔐 Detect SSH service
"""
try:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(3)
sock.connect((self.host, self.port))
banner = sock.recv(1024).decode('utf-8', errors='ignore')
sock.close()
if "SSH" in banner:
return f"SSH 🔐 - {banner.strip()}"
except:
pass
return None
def detect_smtp(self):
"""
# 📧 Detect SMTP service
"""
try:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(3)
sock.connect((self.host, self.port))
banner = sock.recv(1024).decode('utf-8', errors='ignore')
if "220" in banner or "SMTP" in banner.upper():
sock.send(b"EHLO test\r\n")
response = sock.recv(1024).decode('utf-8', errors='ignore')
sock.close()
return f"SMTP 📧 - {banner.split('\r\n')[0]}"
except:
pass
return None
def detect_service(self):
"""
# 🎯 Auto-detect service type
"""
detectors = [
self.detect_http,
self.detect_ssh,
self.detect_smtp
]
for detector in detectors:
result = detector()
if result:
return result
return "Unknown Service ❓"
# 🎮 Test service detection
detector = ServiceDetector("scanme.nmap.org", 22)
print(f"🔍 Service detected: {detector.detect_service()}")⚠️ Common Pitfalls and Solutions
😱 Pitfall 1: Too Aggressive Scanning
# ❌ Wrong way - too fast, might trigger IDS!
def bad_scanner(host):
for port in range(1, 65536):
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(0.1) # 💥 Too short!
sock.connect_ex((host, port)) # 💥 No delay!
sock.close()
# ✅ Correct way - controlled scanning!
def good_scanner(host, delay=0.1):
import time
import random
for port in range(1, 1025): # 🎯 Reasonable range
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(1) # ✅ Reasonable timeout
sock.connect_ex((host, port))
sock.close()
# ⏱️ Add random delay
time.sleep(delay + random.uniform(0, 0.1))🤯 Pitfall 2: Ignoring Legal/Ethical Concerns
# ❌ Dangerous - scanning without permission!
def unethical_scan(random_ip):
scan_ports(random_ip, 1, 65535) # 💥 Illegal!
# ✅ Safe - always get permission!
def ethical_scan(host):
"""
⚠️ IMPORTANT: Only scan systems you own or have
written permission to test!
"""
print("🔒 Security Notice:")
print("- Only scan your own systems")
print("- Get written permission for others")
print("- Respect rate limits")
print("- Follow responsible disclosure")
# ✅ Use designated test hosts
safe_hosts = [
"scanme.nmap.org", # Public test host
"localhost", # Your machine
"192.168.1.1" # Your router (if you own it)
]
if host in safe_hosts or host.startswith("192.168."):
return True
else:
print("⚠️ Please ensure you have permission!")
return False🛠️ Best Practices
- 🎯 Always Get Permission: Only scan systems you own or have written authorization to test!
- 📝 Log Everything: Keep detailed logs of your scanning activities
- 🛡️ Use Rate Limiting: Don’t overwhelm target systems with requests
- 🎨 Be Stealthy When Needed: Use SYN scanning for minimal footprint
- ✨ Verify Results: Double-check findings with different methods
🧪 Hands-On Exercise
🎯 Challenge: Build a Network Mapper
Create a comprehensive network mapping tool:
📋 Requirements:
- ✅ Discover active hosts in a subnet
- 🏷️ Identify open ports on each host
- 👤 Detect running services
- 📅 Generate a network map report
- 🎨 Support different scanning techniques!
🚀 Bonus Points:
- Add OS fingerprinting
- Implement vulnerability detection
- Create a GUI interface
💡 Solution
🔍 Click to see solution
# 🎯 Comprehensive network mapper!
import socket
import threading
import ipaddress
from datetime import datetime
import json
class NetworkMapper:
def __init__(self, network):
self.network = ipaddress.ip_network(network, strict=False)
self.results = {}
self.lock = threading.Lock()
def scan_host(self, ip):
"""
# 🖥️ Scan individual host
"""
ip_str = str(ip)
host_info = {
'ip': ip_str,
'hostname': None,
'open_ports': [],
'services': {},
'scan_time': datetime.now().isoformat()
}
# 🏷️ Try to get hostname
try:
host_info['hostname'] = socket.gethostbyaddr(ip_str)[0]
except:
host_info['hostname'] = 'Unknown'
# 🔍 Scan common ports
common_ports = [21, 22, 23, 25, 53, 80, 110, 443, 445, 3306, 3389, 8080]
for port in common_ports:
if self.check_port(ip_str, port):
host_info['open_ports'].append(port)
# 🎨 Identify service
service_info = self.identify_service_advanced(ip_str, port)
host_info['services'][port] = service_info
# 📝 Store results
if host_info['open_ports']:
with self.lock:
self.results[ip_str] = host_info
print(f"✅ Mapped: {ip_str} ({host_info['hostname']})")
print(f" 📊 Open ports: {host_info['open_ports']}")
def check_port(self, host, port):
"""
# 🔌 Check if port is open
"""
try:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(1)
result = sock.connect_ex((host, port))
sock.close()
return result == 0
except:
return False
def identify_service_advanced(self, host, port):
"""
# 🔍 Advanced service identification
"""
service_info = {
'name': identify_service(port),
'banner': '',
'version': ''
}
try:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(2)
sock.connect((host, port))
# 👋 Send probe based on port
if port == 80 or port == 8080:
sock.send(b"GET / HTTP/1.0\r\n\r\n")
elif port == 22:
pass # SSH sends banner automatically
else:
sock.send(b"\r\n")
# 📋 Receive response
banner = sock.recv(1024).decode('utf-8', errors='ignore')
sock.close()
service_info['banner'] = banner[:100].strip()
# 🎯 Extract version info
if "SSH" in banner:
service_info['version'] = banner.split()[0]
elif "HTTP" in banner:
for line in banner.split('\r\n'):
if 'Server:' in line:
service_info['version'] = line.split(':', 1)[1].strip()
break
except:
pass
return service_info
def map_network(self, max_threads=50):
"""
# 🌐 Map entire network
"""
print(f"🗺️ Starting network mapping for {self.network}")
print(f"📊 Hosts to scan: {self.network.num_addresses - 2}\n")
start_time = datetime.now()
# 🚀 Thread pool for concurrent scanning
threads = []
hosts = list(self.network.hosts())
for i in range(0, len(hosts), max_threads):
batch = hosts[i:i + max_threads]
batch_threads = []
for host in batch:
t = threading.Thread(target=self.scan_host, args=(host,))
t.start()
batch_threads.append(t)
# ⏳ Wait for batch to complete
for t in batch_threads:
t.join()
# 📊 Generate report
duration = datetime.now() - start_time
self.generate_report(duration)
def generate_report(self, duration):
"""
# 📋 Generate network map report
"""
print(f"\n{'='*60}")
print(f"🗺️ Network Mapping Complete!")
print(f"⏱️ Duration: {duration}")
print(f"🌐 Network: {self.network}")
print(f"✅ Active hosts: {len(self.results)}")
print(f"{'='*60}\n")
# 📊 Summary by service
service_count = {}
for host_info in self.results.values():
for port, service in host_info['services'].items():
service_name = service['name']
service_count[service_name] = service_count.get(service_name, 0) + 1
print("📊 Services Summary:")
for service, count in sorted(service_count.items(), key=lambda x: x[1], reverse=True):
print(f" {service}: {count} host(s)")
# 💾 Save detailed report
report_file = f"network_map_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
with open(report_file, 'w') as f:
json.dump(self.results, f, indent=2)
print(f"\n💾 Detailed report saved to: {report_file}")
# 🎮 Map your network!
mapper = NetworkMapper("192.168.1.0/24")
mapper.map_network()🎓 Key Takeaways
You’ve learned so much! Here’s what you can now do:
- ✅ Perform port scanning with confidence 💪
- ✅ Discover network services like a pro 🛡️
- ✅ Apply security best practices in your scans 🎯
- ✅ Debug network issues efficiently 🐛
- ✅ Build network discovery tools with Python! 🚀
Remember: With great power comes great responsibility! Always scan ethically and legally. 🤝
🤝 Next Steps
Congratulations! 🎉 You’ve mastered port scanning and network discovery!
Here’s what to do next:
- 💻 Practice with the exercises above on authorized systems
- 🏗️ Build a network monitoring dashboard
- 📚 Move on to our next tutorial: UDP Sockets and Connectionless Communication
- 🌟 Share your network discoveries responsibly!
Remember: Every network security expert started by learning the fundamentals. Keep exploring, keep learning, and most importantly, stay ethical! 🚀
Happy network discovering! 🎉🚀✨