Skip to main content

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