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",
        )
pip install streamlit pandas psutil
streamlit run scanner.py
http://localhost:8501
Migrated from Apple Server Wiki. Original page ID: z3L1C1o972