mirror of
https://github.com/wahyd4/links.git
synced 2026-08-08 21:04:53 +10:00
118 lines
3.8 KiB
Python
118 lines
3.8 KiB
Python
import socket
|
|
import logging
|
|
from .base import Finding
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
CHECK_NAME = 'router_ports'
|
|
|
|
PORTS_TO_PROBE = [22, 23, 53, 80, 139, 443, 445, 8080, 8443]
|
|
TIMEOUT = 3
|
|
|
|
|
|
def _tcp_open(host: str, port: int) -> bool:
|
|
try:
|
|
with socket.create_connection((host, port), timeout=TIMEOUT):
|
|
return True
|
|
except (OSError, socket.timeout):
|
|
return False
|
|
|
|
|
|
def _fetch_http_headers(host: str, port: int = 80) -> dict:
|
|
try:
|
|
with socket.create_connection((host, port), timeout=TIMEOUT) as s:
|
|
s.sendall(f'HEAD / HTTP/1.0\r\nHost: {host}\r\n\r\n'.encode())
|
|
resp = s.recv(4096).decode('utf-8', errors='replace')
|
|
headers = {}
|
|
for line in resp.splitlines()[1:]:
|
|
if ':' in line:
|
|
k, _, v = line.partition(':')
|
|
headers[k.strip().lower()] = v.strip()
|
|
return headers
|
|
except Exception:
|
|
return {}
|
|
|
|
|
|
def run(profile) -> list:
|
|
findings = []
|
|
host = profile.gateway_ip
|
|
open_ports = {}
|
|
|
|
for port in PORTS_TO_PROBE:
|
|
open_ports[port] = _tcp_open(host, port)
|
|
|
|
raw = {'gateway_ip': host, 'open_ports': {str(p): v for p, v in open_ports.items()}}
|
|
|
|
# SMB exposure
|
|
if open_ports.get(139) or open_ports.get(445):
|
|
smb_ports = [p for p in [139, 445] if open_ports.get(p)]
|
|
findings.append(Finding(
|
|
check_name=CHECK_NAME,
|
|
severity='warning',
|
|
title=f'SMB ports open on gateway ({", ".join(str(p) for p in smb_ports)})',
|
|
detail='Windows file sharing (SMB) is accessible on the gateway. This could expose network shares.',
|
|
raw=raw,
|
|
))
|
|
|
|
# Telnet
|
|
if open_ports.get(23):
|
|
findings.append(Finding(
|
|
check_name=CHECK_NAME,
|
|
severity='critical',
|
|
title='Telnet port 23 open on gateway',
|
|
detail='Telnet transmits credentials in plaintext. Disable telnet and use SSH instead.',
|
|
raw=raw,
|
|
))
|
|
|
|
# SSH open on gateway
|
|
if open_ports.get(22):
|
|
findings.append(Finding(
|
|
check_name=CHECK_NAME,
|
|
severity='warning',
|
|
title='SSH port 22 open on gateway',
|
|
detail='SSH is accessible on the gateway. Ensure key-only auth is enforced and access is restricted.',
|
|
raw=raw,
|
|
))
|
|
|
|
# Plain HTTP admin (port 80 open, port 443 closed)
|
|
if open_ports.get(80) and not open_ports.get(443):
|
|
headers = _fetch_http_headers(host, 80)
|
|
findings.append(Finding(
|
|
check_name=CHECK_NAME,
|
|
severity='warning',
|
|
title='Gateway admin over plain HTTP (no HTTPS)',
|
|
detail=f'Port 80 is open but 443 is closed. Admin interface may be served unencrypted. Server header: {headers.get("server", "unknown")}',
|
|
raw={**raw, 'http_headers': headers},
|
|
))
|
|
|
|
# Unknown port 8080
|
|
if open_ports.get(8080):
|
|
findings.append(Finding(
|
|
check_name=CHECK_NAME,
|
|
severity='info',
|
|
title='Port 8080 open on gateway',
|
|
detail='An alternate HTTP service is running on port 8080. Verify this is intentional.',
|
|
raw=raw,
|
|
))
|
|
|
|
# Port 8443 open
|
|
if open_ports.get(8443):
|
|
findings.append(Finding(
|
|
check_name=CHECK_NAME,
|
|
severity='info',
|
|
title='Port 8443 open on gateway',
|
|
detail='An alternate HTTPS service is running on port 8443. Verify this is intentional.',
|
|
raw=raw,
|
|
))
|
|
|
|
if not findings:
|
|
findings.append(Finding(
|
|
check_name=CHECK_NAME,
|
|
severity='ok',
|
|
title='Gateway port scan looks clean',
|
|
detail=f'No high-risk ports found open on {host}.',
|
|
raw=raw,
|
|
))
|
|
|
|
return findings
|