python scanner project
sudo nano scanner.py
# network_scanner.py
import ipaddress
import platform
import socket
import subprocess
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime
import pandas as pd
import psutil
import streamlit as st
COMMON_PORTS = {
22: "SSH",
53: "DNS",
80: "HTTP",
443: "HTTPS",
445: "SMB",
3389: "RDP",
5900: "VNC",
8080: "HTTP-ALT",
}
def get_local_ip():
"""
Determine the primary local IPv4 address.
"""
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
try:
s.connect(("8.8.8.8", 80))
ip = s.getsockname()[0]
except Exception:
ip = "127.0.0.1"
finally:
s.close()
return ip
def guess_network():
"""
Try to find the subnet/netmask associated with the primary interface.
"""
local_ip = get_local_ip()
for interface, addresses in psutil.net_if_addrs().items():
for addr in addresses:
if addr.family == socket.AF_INET and addr.address == local_ip:
if addr.netmask:
network = ipaddress.IPv4Network(
f"{addr.address}/{addr.netmask}",
strict=False
)
return str(network)
# fallback
return f"{local_ip}/24"
def ping_host(ip):
"""
Ping host once.
"""
system = platform.system().lower()
if system == "windows":
command = ["ping", "-n", "1", "-w", "700", ip]
else:
command = ["ping", "-c", "1", "-W", "1", ip]
try:
result = subprocess.run(
command,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
timeout=2,
)
return result.returncode == 0
except Exception:
return False
def resolve_hostname(ip):
try:
hostname, _, _ = socket.gethostbyaddr(ip)
return hostname
except Exception:
return ""
def check_port(ip, port, timeout=0.25):
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(timeout)
try:
return sock.connect_ex((ip, port)) == 0
except Exception:
return False
finally:
sock.close()
def scan_ports(ip):
open_ports = []
for port, service in COMMON_PORTS.items():
if check_port(ip, port):
open_ports.append(f"{port}/{service}")
return ", ".join(open_ports)
def scan_host(ip):
ip = str(ip)
if not ping_host(ip):
return None
hostname = resolve_hostname(ip)
ports = scan_ports(ip)
return {
"IP Address": ip,
"Hostname": hostname,
"Open Ports": ports,
"Status": "Online",
}
def scan_network(network_string, workers=64):
network = ipaddress.ip_network(network_string, strict=False)
hosts = list(network.hosts())
results = []
progress = st.progress(0)
status = st.empty()
completed = 0
with ThreadPoolExecutor(max_workers=workers) as executor:
futures = {
executor.submit(scan_host, host): host
for host in hosts
}
for future in as_completed(futures):
completed += 1
try:
result = future.result()
if result:
results.append(result)
except Exception:
pass
progress.progress(completed / len(hosts))
status.text(
f"Scanning {completed}/{len(hosts)} addresses..."
)
progress.empty()
status.empty()
return results
# ------------------------------------------------------
# STREAMLIT DASHBOARD
# ------------------------------------------------------
st.set_page_config(
page_title="Network Scanner",
page_icon="🌐",
layout="wide",
)
st.title("🌐 Network Scanner")
st.caption(
"Discover active hosts and common network services on an authorized network."
)
default_network = guess_network()
with st.sidebar:
st.header("Scan Settings")
network_input = st.text_input(
"Network / CIDR",
value=default_network,
help="Example: 192.168.1.0/24",
)
workers = st.slider(
"Concurrent workers",
min_value=8,
max_value=128,
value=64,
step=8,
)
scan_button = st.button(
"Start Scan",
type="primary",
use_container_width=True,
)
st.divider()
st.write("Local IP")
st.code(get_local_ip())
st.write("Detected network")
st.code(default_network)
if scan_button:
try:
network = ipaddress.ip_network(
network_input,
strict=False
)
if network.num_addresses > 1024:
st.warning(
"This dashboard limits scans to networks of "
"1024 addresses or fewer."
)
st.stop()
except ValueError:
st.error("Invalid network. Example: 192.168.1.0/24")
st.stop()
st.info(f"Scanning {network_input}")
start = datetime.now()
results = scan_network(
network_input,
workers=workers
)
elapsed = (datetime.now() - start).total_seconds()
st.session_state["results"] = results
st.success(
f"Scan completed in {elapsed:.1f} seconds."
)
if "results" in st.session_state:
results = st.session_state["results"]
if not results:
st.warning("No active hosts found.")
else:
df = pd.DataFrame(results)
# sort numerically by IP
df["_ip_sort"] = df["IP Address"].apply(
lambda x: int(ipaddress.ip_address(x))
)
df = (
df.sort_values("_ip_sort")
.drop(columns=["_ip_sort"])
)
total_hosts = len(df)
hosts_with_ports = (
df["Open Ports"]
.astype(bool)
.sum()
)
c1, c2, c3 = st.columns(3)
c1.metric(
"Online Hosts",
total_hosts
)
c2.metric(
"Hosts With Open Ports",
hosts_with_ports
)
c3.metric(
"Network",
network_input
)
st.subheader("Discovered Devices")
st.dataframe(
df,
use_container_width=True,
hide_index=True,
)
csv = df.to_csv(
index=False
).encode("utf-8")
st.download_button(
"Download CSV",
csv,
"network_scan.csv",
"text/csv",
)
import ipaddress
import platform
import socket
import subprocess
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime
import pandas as pd
import psutil
import streamlit as st
COMMON_PORTS = {
22: "SSH",
53: "DNS",
80: "HTTP",
443: "HTTPS",
445: "SMB",
3389: "RDP",
5900: "VNC",
8080: "HTTP-ALT",
}
def get_local_ip():
"""
Determine the primary local IPv4 address.
"""
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
try:
s.connect(("8.8.8.8", 80))
ip = s.getsockname()[0]
except Exception:
ip = "127.0.0.1"
finally:
s.close()
return ip
def guess_network():
"""
Try to find the subnet/netmask associated with the primary interface.
"""
local_ip = get_local_ip()
for interface, addresses in psutil.net_if_addrs().items():
for addr in addresses:
if addr.family == socket.AF_INET and addr.address == local_ip:
if addr.netmask:
network = ipaddress.IPv4Network(
f"{addr.address}/{addr.netmask}",
strict=False
)
return str(network)
# fallback
return f"{local_ip}/24"
def ping_host(ip):
"""
Ping host once.
"""
system = platform.system().lower()
if system == "windows":
command = ["ping", "-n", "1", "-w", "700", ip]
else:
command = ["ping", "-c", "1", "-W", "1", ip]
try:
result = subprocess.run(
command,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
timeout=2,
)
return result.returncode == 0
except Exception:
return False
def resolve_hostname(ip):
try:
hostname, _, _ = socket.gethostbyaddr(ip)
return hostname
except Exception:
return ""
def check_port(ip, port, timeout=0.25):
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(timeout)
try:
return sock.connect_ex((ip, port)) == 0
except Exception:
return False
finally:
sock.close()
def scan_ports(ip):
open_ports = []
for port, service in COMMON_PORTS.items():
if check_port(ip, port):
open_ports.append(f"{port}/{service}")
return ", ".join(open_ports)
def scan_host(ip):
ip = str(ip)
if not ping_host(ip):
return None
hostname = resolve_hostname(ip)
ports = scan_ports(ip)
return {
"IP Address": ip,
"Hostname": hostname,
"Open Ports": ports,
"Status": "Online",
}
def scan_network(network_string, workers=64):
network = ipaddress.ip_network(network_string, strict=False)
hosts = list(network.hosts())
results = []
progress = st.progress(0)
status = st.empty()
completed = 0
with ThreadPoolExecutor(max_workers=workers) as executor:
futures = {
executor.submit(scan_host, host): host
for host in hosts
}
for future in as_completed(futures):
completed += 1
try:
result = future.result()
if result:
results.append(result)
except Exception:
pass
progress.progress(completed / len(hosts))
status.text(
f"Scanning {completed}/{len(hosts)} addresses..."
)
progress.empty()
status.empty()
return results
# ------------------------------------------------------
# STREAMLIT DASHBOARD
# ------------------------------------------------------
st.set_page_config(
page_title="Network Scanner",
page_icon="🌐",
layout="wide",
)
st.title("🌐 Network Scanner")
st.caption(
"Discover active hosts and common network services on an authorized network."
)
default_network = guess_network()
with st.sidebar:
st.header("Scan Settings")
network_input = st.text_input(
"Network / CIDR",
value=default_network,
help="Example: 192.168.1.0/24",
)
workers = st.slider(
"Concurrent workers",
min_value=8,
max_value=128,
value=64,
step=8,
)
scan_button = st.button(
"Start Scan",
type="primary",
use_container_width=True,
)
st.divider()
st.write("Local IP")
st.code(get_local_ip())
st.write("Detected network")
st.code(default_network)
if scan_button:
try:
network = ipaddress.ip_network(
network_input,
strict=False
)
if network.num_addresses > 1024:
st.warning(
"This dashboard limits scans to networks of "
"1024 addresses or fewer."
)
st.stop()
except ValueError:
st.error("Invalid network. Example: 192.168.1.0/24")
st.stop()
st.info(f"Scanning {network_input}")
start = datetime.now()
results = scan_network(
network_input,
workers=workers
)
elapsed = (datetime.now() - start).total_seconds()
st.session_state["results"] = results
st.success(
f"Scan completed in {elapsed:.1f} seconds."
)
if "results" in st.session_state:
results = st.session_state["results"]
if not results:
st.warning("No active hosts found.")
else:
df = pd.DataFrame(results)
# sort numerically by IP
df["_ip_sort"] = df["IP Address"].apply(
lambda x: int(ipaddress.ip_address(x))
)
df = (
df.sort_values("_ip_sort")
.drop(columns=["_ip_sort"])
)
total_hosts = len(df)
hosts_with_ports = (
df["Open Ports"]
.astype(bool)
.sum()
)
c1, c2, c3 = st.columns(3)
c1.metric(
"Online Hosts",
total_hosts
)
c2.metric(
"Hosts With Open Ports",
hosts_with_ports
)
c3.metric(
"Network",
network_input
)
st.subheader("Discovered Devices")
st.dataframe(
df,
use_container_width=True,
hide_index=True,
)
csv = df.to_csv(
index=False
).encode("utf-8")
st.download_button(
"Download CSV",
csv,
"network_scan.csv",
"text/csv",
)
pip install streamlit pandas psutil
streamlit run scanner.py
http://localhost:8501
Migrated from Apple Server Wiki. Original page ID: z3L1C1o972