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 = 'camera_rtsp'
|
|
TIMEOUT = 5
|
|
RTSP_PORT = 554
|
|
|
|
|
|
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 _rtsp_request(host: str, method: str, seq: int, extra_headers: str = '') -> str:
|
|
try:
|
|
with socket.create_connection((host, RTSP_PORT), timeout=TIMEOUT) as s:
|
|
request = (
|
|
f'{method} rtsp://{host}/ RTSP/1.0\r\n'
|
|
f'CSeq: {seq}\r\n'
|
|
f'{extra_headers}'
|
|
'\r\n'
|
|
)
|
|
s.sendall(request.encode())
|
|
response = s.recv(4096).decode('utf-8', errors='replace')
|
|
return response
|
|
except Exception as e:
|
|
return f'ERROR: {e}'
|
|
|
|
|
|
def _parse_rtsp_status(response: str) -> int:
|
|
"""Extract HTTP-style status code from RTSP response."""
|
|
try:
|
|
first_line = response.splitlines()[0]
|
|
return int(first_line.split()[1])
|
|
except (IndexError, ValueError):
|
|
return 0
|
|
|
|
|
|
def run(profile) -> list:
|
|
findings = []
|
|
cameras = profile.cameras or []
|
|
|
|
if not cameras:
|
|
return [Finding(
|
|
check_name=CHECK_NAME,
|
|
severity='info',
|
|
title='No camera IPs configured',
|
|
detail='Add camera IPs to the scan profile to enable RTSP unauthenticated access check.',
|
|
raw={},
|
|
)]
|
|
|
|
for ip in cameras:
|
|
raw = {'camera_ip': ip}
|
|
|
|
if not _tcp_open(ip, RTSP_PORT):
|
|
findings.append(Finding(
|
|
check_name=CHECK_NAME,
|
|
severity='info',
|
|
title=f'{ip}: RTSP port 554 closed',
|
|
detail=f'Port 554 is not open on {ip}. Camera may be offline or not using RTSP.',
|
|
raw=raw,
|
|
))
|
|
continue
|
|
|
|
options_resp = _rtsp_request(ip, 'OPTIONS', 1)
|
|
raw['options_response'] = options_resp[:500]
|
|
options_status = _parse_rtsp_status(options_resp)
|
|
|
|
if options_status == 0:
|
|
findings.append(Finding(
|
|
check_name=CHECK_NAME,
|
|
severity='info',
|
|
title=f'{ip}: RTSP OPTIONS failed',
|
|
detail=f'Got unexpected RTSP OPTIONS response from {ip}.',
|
|
raw=raw,
|
|
))
|
|
continue
|
|
|
|
describe_resp = _rtsp_request(ip, 'DESCRIBE', 2, 'Accept: application/sdp\r\n')
|
|
raw['describe_response'] = describe_resp[:500]
|
|
describe_status = _parse_rtsp_status(describe_resp)
|
|
|
|
if describe_status == 200:
|
|
findings.append(Finding(
|
|
check_name=CHECK_NAME,
|
|
severity='critical',
|
|
title=f'{ip}: RTSP stream accessible without credentials',
|
|
detail=(
|
|
f'Camera at {ip} returned 200 to DESCRIBE without authentication. '
|
|
'Live stream may be publicly accessible on the LAN.'
|
|
),
|
|
raw=raw,
|
|
))
|
|
elif describe_status in (401, 403):
|
|
findings.append(Finding(
|
|
check_name=CHECK_NAME,
|
|
severity='ok',
|
|
title=f'{ip}: RTSP requires authentication',
|
|
detail=f'Camera at {ip} returned {describe_status} to DESCRIBE — auth is enforced.',
|
|
raw=raw,
|
|
))
|
|
else:
|
|
findings.append(Finding(
|
|
check_name=CHECK_NAME,
|
|
severity='ok',
|
|
title=f'{ip}: RTSP DESCRIBE returned {describe_status}',
|
|
detail=f'Camera at {ip} responded with status {describe_status} — no unauthenticated stream detected.',
|
|
raw=raw,
|
|
))
|
|
|
|
return findings
|