122 lines
4.5 KiB
Python
122 lines
4.5 KiB
Python
"""Isolated ONVIF fixture for Sense acceptance tests.
|
|
|
|
Credentials are required through process environment and are never printed or
|
|
persisted. The fixture validates RFC 7616 MD5 qop=auth so the test exercises
|
|
Sense's real Digest retry instead of accepting an arbitrary header.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from pathlib import Path
|
|
|
|
|
|
HOST = "127.0.0.1"
|
|
PORT = int(os.environ["SENSE_E2E_ONVIF_PORT"])
|
|
RTSP_PORT = int(os.environ["SENSE_E2E_RTSP_PORT"])
|
|
USERNAME = os.environ["SENSE_E2E_CAMERA_USERNAME"]
|
|
PASSWORD = os.environ["SENSE_E2E_CAMERA_PASSWORD"]
|
|
STATE_PATH = Path(os.environ["SENSE_E2E_PROFILE_STATE_FILE"])
|
|
STATUS_PATH = Path(os.environ["SENSE_E2E_FIXTURE_STATUS_FILE"])
|
|
REALM = "sense-e2e-camera"
|
|
NONCE = hashlib.sha256(os.urandom(32)).hexdigest()
|
|
|
|
|
|
def _md5(value: str) -> str:
|
|
return hashlib.md5(value.encode("utf-8"), usedforsecurity=False).hexdigest()
|
|
|
|
|
|
def _parameters(value: str) -> dict[str, str]:
|
|
result: dict[str, str] = {}
|
|
for match in re.finditer(r'(\w+)=(?:"([^"]*)"|([^,\s]+))', value):
|
|
result[match.group(1).lower()] = match.group(2) or match.group(3)
|
|
return result
|
|
|
|
|
|
def _authorized(header: str, method: str, uri: str) -> bool:
|
|
if not header.startswith("Digest "):
|
|
return False
|
|
values = _parameters(header[7:])
|
|
if values.get("username") != USERNAME or values.get("realm") != REALM:
|
|
return False
|
|
if values.get("nonce") != NONCE or values.get("uri") != uri:
|
|
return False
|
|
ha1 = _md5(f"{USERNAME}:{REALM}:{PASSWORD}")
|
|
ha2 = _md5(f"{method}:{uri}")
|
|
if values.get("qop") == "auth":
|
|
expected = _md5(
|
|
f"{ha1}:{NONCE}:{values.get('nc', '')}:{values.get('cnonce', '')}:auth:{ha2}"
|
|
)
|
|
else:
|
|
expected = _md5(f"{ha1}:{NONCE}:{ha2}")
|
|
return values.get("response") == expected
|
|
|
|
|
|
def _write_status(digest_requests: int) -> None:
|
|
STATUS_PATH.write_text(
|
|
json.dumps({"ready": True, "digestRequests": digest_requests}),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
|
|
class Handler(BaseHTTPRequestHandler):
|
|
digest_requests = 0
|
|
|
|
def log_message(self, _format: str, *_args: object) -> None:
|
|
return
|
|
|
|
def do_POST(self) -> None: # noqa: N802 - BaseHTTPRequestHandler API
|
|
length = min(int(self.headers.get("Content-Length", "0")), 2 << 20)
|
|
body = self.rfile.read(length).decode("utf-8", errors="replace")
|
|
authorization = self.headers.get("Authorization", "")
|
|
if not _authorized(authorization, "POST", self.path):
|
|
self.send_response(401)
|
|
self.send_header(
|
|
"WWW-Authenticate",
|
|
f'Digest realm="{REALM}", nonce="{NONCE}", algorithm=MD5, qop="auth"',
|
|
)
|
|
self.end_headers()
|
|
return
|
|
|
|
type(self).digest_requests += 1
|
|
_write_status(type(self).digest_requests)
|
|
state = STATE_PATH.read_text(encoding="utf-8").strip()
|
|
width, height = ((1280, 720) if state == "changed" else (1920, 1080))
|
|
if "GetCapabilities" in body:
|
|
payload = (
|
|
"<Envelope><Body><GetCapabilitiesResponse><Capabilities><Media>"
|
|
f"<XAddr>http://{HOST}:{PORT}/onvif/media</XAddr>"
|
|
"</Media></Capabilities></GetCapabilitiesResponse></Body></Envelope>"
|
|
)
|
|
elif "GetProfiles" in body:
|
|
payload = (
|
|
'<Envelope><Body><GetProfilesResponse><Profiles token="main">'
|
|
"<Name>主码流</Name><VideoEncoderConfiguration><Encoding>H264</Encoding>"
|
|
f"<Resolution><Width>{width}</Width><Height>{height}</Height></Resolution>"
|
|
"</VideoEncoderConfiguration></Profiles></GetProfilesResponse></Body></Envelope>"
|
|
)
|
|
elif "GetStreamUri" in body:
|
|
payload = (
|
|
"<Envelope><Body><GetStreamUriResponse><MediaUri>"
|
|
f"<Uri>rtsp://{HOST}:{RTSP_PORT}/fixture</Uri>"
|
|
"</MediaUri></GetStreamUriResponse></Body></Envelope>"
|
|
)
|
|
else:
|
|
self.send_error(400)
|
|
return
|
|
encoded = payload.encode("utf-8")
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", "application/soap+xml; charset=utf-8")
|
|
self.send_header("Content-Length", str(len(encoded)))
|
|
self.end_headers()
|
|
self.wfile.write(encoded)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
_write_status(0)
|
|
ThreadingHTTPServer((HOST, PORT), Handler).serve_forever()
|