WebAVP/app.py

#!/usr/bin/env python3
"""Alta Video Player — lightweight HTTP server using only Python stdlib."""

import json
import mimetypes
import os
from http.server import HTTPServer, SimpleHTTPRequestHandler
from urllib.parse import parse_qs, urlencode, urlparse
from urllib.request import urlopen
from urllib.error import HTTPError, URLError

# Loopback is the safe default for a local footage player. Set WEBAVP_HOST
# explicitly only when sharing on a trusted network.
HOST = os.environ.get("WEBAVP_HOST", "127.0.0.1")
PORT = int(os.environ.get("WEBAVP_PORT", "5152"))
LATEST_RELEASE_URL = "https://git.pejicorp.com/api/v1/repos/peji/WebAVP/releases/latest"
MAX_RELEASE_RESPONSE_BYTES = 1024 * 1024
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
STATIC_DIR = os.path.join(BASE_DIR, "static")
TEMPLATES_DIR = os.path.join(BASE_DIR, "templates")


class Handler(SimpleHTTPRequestHandler):
    def do_GET(self):
        parsed = urlparse(self.path)
        path = parsed.path

        if path == "/":
            self._serve_file(os.path.join(TEMPLATES_DIR, "index.html"), "text/html")
        elif path.startswith("/static/"):
            rel = path[len("/static/"):]
            file_path = os.path.join(STATIC_DIR, rel)
            static_root = os.path.realpath(STATIC_DIR)
            requested_path = os.path.realpath(file_path)
            if requested_path != static_root and not requested_path.startswith(static_root + os.sep):
                self.send_error(403)
                return
            mime = mimetypes.guess_type(file_path)[0] or "application/octet-stream"
            self._serve_file(file_path, mime)
        elif path == "/api/verify-cert":
            self._proxy_verify(parsed.query)
        elif path == "/api/latest-release":
            self._proxy_latest_release()
        else:
            self.send_error(404)

    def _serve_file(self, file_path, content_type):
        try:
            with open(file_path, "rb") as f:
                data = f.read()
            self.send_response(200)
            self.send_header("Content-Type", content_type)
            self.send_header("Content-Length", len(data))
            self.end_headers()
            self.wfile.write(data)
        except FileNotFoundError:
            self.send_error(404)

    def _proxy_verify(self, query_string):
        params = parse_qs(query_string)
        serial = params.get("serial", [""])[0]
        cert_hash = params.get("certificateHash", [""])[0]

        if not serial or not cert_hash:
            self._json_response(400, {"verified": False, "error": "Missing parameters"})
            return

        try:
            qs = urlencode({"serial": serial, "certificateHash": cert_hash})
            url = f"https://aware.avasecurity.com/api/v1/public/verifyServerCertificate?{qs}"
            resp = urlopen(url, timeout=10)
            body = resp.read()
            if not 200 <= resp.status < 300:
                self._json_response(200, {"verified": False, "error": f"HTTP {resp.status}"})
            elif self._interpret_verify_body(body) is False:
                self._json_response(200, {"verified": False, "error": "Certificate not recognized by Alta/Ava"})
            else:
                self._json_response(200, {"verified": True})
        except HTTPError as e:
            self._json_response(200, {"verified": False, "error": f"HTTP {e.code}"})
        except URLError as e:
            self._json_response(200, {"verified": False, "error": str(e)})

    def _proxy_latest_release(self):
        """Same-origin, credential-free proxy for the public GitPeji release."""
        try:
            response = urlopen(LATEST_RELEASE_URL, timeout=10)
            body = response.read(MAX_RELEASE_RESPONSE_BYTES + 1)
            if len(body) > MAX_RELEASE_RESPONSE_BYTES:
                self._json_response(502, {"error": "GitPeji release response exceeded size limit"})
                return
            self._raw_response(response.status, body, "application/json; charset=utf-8")
        except HTTPError as e:
            body = e.read(MAX_RELEASE_RESPONSE_BYTES + 1)
            if len(body) > MAX_RELEASE_RESPONSE_BYTES:
                body = b'{"error":"GitPeji release response exceeded size limit"}'
            self._raw_response(e.code, body, "application/json; charset=utf-8")
        except URLError as e:
            self._json_response(502, {"error": str(e)})

    @staticmethod
    def _interpret_verify_body(body):
        """Return an explicit upstream result, or None for unknown schemas."""
        text = body.decode("utf-8", "replace").strip() if body else ""
        if not text:
            return None
        lowered = text.lower()
        if lowered in ("true", "false"):
            return lowered == "true"
        try:
            data = json.loads(text)
        except (ValueError, TypeError):
            return None
        if isinstance(data, bool):
            return data
        if isinstance(data, dict):
            for field in ("verified", "valid", "isValid", "success", "result"):
                if isinstance(data.get(field), bool):
                    return data[field]
        return None

    def _json_response(self, status, data):
        body = json.dumps(data).encode()
        self._raw_response(status, body, "application/json")

    def _raw_response(self, status, body, content_type):
        self.send_response(status)
        self.send_header("Content-Type", content_type)
        self.send_header("Content-Length", len(body))
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, fmt, *args):
        print(f"[AVP] {args[0]}")


if __name__ == "__main__":
    server = HTTPServer((HOST, PORT), Handler)
    print(f"Alta Video Player running on http://{HOST}:{PORT}")
    server.serve_forever()