update

.github/workflows/CI.yaml

@@ -11,12 +11,12 @@ jobs:
       - name: Set up Python 3.12
         uses: actions/setup-python@v4
         with:
-          python-version: '3.12'
+          python-version: "3.12"
       - name: Install dependencies
         run: pip install -r requirements.txt
       - name: Run tests and collect coverage
         run: pytest --cov src ${{ env.CODECOV_ATS_TESTS }}
       - name: Upload coverage reports to Codecov
         uses: codecov/[email protected]
         with:
-          token: ${{ secrets.CODECOV_TOKEN }}
+          token: ${{ secrets.CODECOV_TOKEN }}

.gitignore

@@ -3,3 +3,5 @@ __pycache__/
 /.venv/
 /.pytest_cache/
 .coverage
+/config.yaml
+/db.json

config.example.yaml

@@ -0,0 +1,62 @@
+# The document template is enabled for this yaml config file. Which can be accessed as examples below:
+# {{browser.headers.Accept}} -> text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8
+# {{epg.authenticator.auth_method}} -> SALTED_MD5
+browser:
+    args:
+        - --disable-extensions
+        - --disable-gpu
+        - --disable-dev-shm-usage
+        - --disable-add-to-shelf
+        - --disable-background-networking
+        - --disable-background-timer-throttling
+        - --disable-backgrounding-occluded-windows
+        - --disable-breakpad
+        - --disable-checker-imaging
+        - --disable-datasaver-prompt
+        - --disable-default-apps
+        - --disable-desktop-notifications
+        - --disable-domain-reliability
+        - --disable-hang-monitor
+        - --disable-infobars
+        - --disable-logging
+        - --disable-notifications
+        - --disable-popup-blocking
+        - --disable-prompt-on-repost
+        - --disable-renderer-backgrounding
+        - --disable-sync
+        - --force-color-profile=srgb
+        - --force-device-scale-factor=1
+        - --metrics-recording-only
+        - --mute-audio
+        - --no-default-browser-check
+        - --no-first-run
+    headers:
+        Accept: text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8
+        Accept-Charset: utf-8, iso-8859-1, utf-16, *;q=0.7
+        Accept-Encoding: gzip
+        Accept-Language: en-us
+        User-Agent:
+            Mozilla/4.0 (compatible; MSIE 8.0; Windows NT 5.1; Trident/4.0)
+    # start_url: Endpoints where the authentication start
+    start_url: "http://127.0.0.1/login?UserID={{epg.credential.user_id}}"
+    # end_url: Endpoints for which configuration files have been distributed, wildcard is allowed to use
+    end_url: "**/everything_finish.html"
+epg:
+    authenticator:
+        # auth_method(PLAIN, MD5, SALTED_MD5): The hashing algorithm your carrier use
+        auth_method: SALTED_MD5
+        # salt[optional], default None: Only need if you are using SALTED_MD5 as hashing algorithm, THIS FIELD MUST BE STRING!
+        salt: "12345678"
+    credential:
+        # user_id: Account
+        user_id: acc_example
+        # password: Password
+        password: pass1234
+        # ip: IP address of your STB, might not affect authentication since the IP is unstable
+        ip: 127.0.0.1
+        # mac: Mac address of your STB, affect authentication
+        mac: 1A:2B:3C:4D:5E:6F
+        # product_id: Unique id of your STB, affect authentication
+        product_id: PRODUCTID1234567890
+        # ctc[optional], default to CTC: Unknown, might affect authentication, I don't know what it means for god sake
+        ctc: SOMECTC

requirements.txt

@@ -1,7 +1,8 @@
-aiohttp>=0.0.0
 fastapi>=0.0.0
-playwright>=0.0.0
+jinja2>=0.0.0
+playwright==1.44.0
 pycryptodome>=0.0.0
+pydantic>=0.0.0
 uvicorn[standard]>=0.0.0
 pytest>=0.0.0
 pytest-cov>=0.0.0

setup.py

@@ -0,0 +1,66 @@
+#!/usr/bin/env python3
+import os
+import subprocess
+import venv
+from pathlib import Path
+
+VENV_PATH = ".venv"
+
+
+def get_binary_path(path: str | Path) -> Path:
+    if not isinstance(path, Path):
+        path = Path(path)
+    for bin_dir in ["bin", "Scripts"]:
+        binary_path = path / bin_dir
+        if binary_path.exists():
+            return binary_path
+    raise FileNotFoundError("binary path of venv not found")
+
+
+def rm(path: str | Path) -> None:
+    if not isinstance(path, Path):
+        path = Path(path)
+    if path.exists():
+        if path.is_dir() and not path.is_symlink():
+            for sub_path in path.iterdir():
+                rm(sub_path)
+            path.rmdir()
+        else:
+            path.unlink()
+    else:
+        raise FileNotFoundError()
+
+
+def create_venv(*args, **kwargs) -> None:
+    path = Path(kwargs["env_dir"] if "env_dir" in kwargs else args[0])
+    print(path)
+    try:
+        rm(path)
+    except FileNotFoundError:
+        pass
+    venv.create(*args, **kwargs)
+
+
+if __name__ != "__main__":
+    raise NotImplementedError()
+
+print("setup venv")
+create_venv(
+    VENV_PATH,
+    with_pip=True,
+    upgrade_deps=True,
+)
+binary_path = get_binary_path(VENV_PATH)
+print("install packages")
+subprocess.run(
+    [binary_path / "pip", "install", "-r", "requirements.txt"], check=True
+)
+print("install browser")
+subprocess.run([binary_path / "playwright", "install", "chromium"], check=True)
+match os.name:
+    case "nt":
+        os.system(f"mklink {binary_path}/")
+    case "posix":
+        os.system(f"ln -s {binary_path}/")
+    case _:
+        raise NotImplementedError()

src/browser/__init__.py

@@ -0,0 +1 @@
+from .browser import process

src/browser/browser.py

@@ -0,0 +1,31 @@
+from playwright.async_api import (
+    Browser,
+    Page,
+    async_playwright,
+)
+
+from . import remote_injector
+
+
+async def process(
+    injector: remote_injector.Injector,
+    start_url: str,
+    end_url: str,
+    args: list | None = None,
+    headers: dict | None = None,
+    headless: bool = True,
+):
+    async with async_playwright() as playwright:
+        browser: Browser = await playwright.chromium.launch(
+            args=args, headless=headless
+        )
+        page: Page = await browser.new_page()
+
+        await remote_injector.inject(page, injector)
+        if headers:
+            await page.set_extra_http_headers(headers)
+
+        await page.goto(start_url)
+
+        await page.wait_for_url(url=end_url, wait_until="domcontentloaded")
+        await browser.close()

src/browser/remote_injector/__init__.py

@@ -0,0 +1,2 @@
+from .inject import inject
+from .injector import Injector

src/browser/remote_injector/inject.py

@@ -0,0 +1,109 @@
+import asyncio
+import inspect
+import socket
+from pathlib import Path
+
+import uvicorn
+from jinja2 import Environment
+from playwright.async_api import (
+    BrowserContext,
+    Page,
+)
+
+from .injector import Injector
+from .server import RemoteInvokeServer
+
+
+def _get_free_port() -> int:
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+        s.bind(("", 0))
+        port = s.getsockname()[1]
+        return port
+
+
+def _generate_script(injector: Injector) -> str:
+    result = ""
+    for class_name, class_ in injector:
+        if class_name != "_":
+            result += f"class {class_name} {{}}\n"
+        for method_name, method in class_.items():
+            if class_name == "_":
+                result += (
+                    f"{method_name} = "
+                    f"function({','.join(inspect.signature(method).parameters)})"
+                    "{return "
+                    f'invokeRemoteFunction("{method_name}",'
+                    "null,"
+                    f'{{{",".join(
+                        f'"{para}":{para}' for para
+                        in inspect.signature(method).parameters
+                    )}}}'
+                    ").result};"
+                    "\n"
+                )
+            else:
+                result += (
+                    f"{class_name}.{method_name} = "
+                    f"function({','.join(inspect.signature(method).parameters)})"
+                    "{return "
+                    f'invokeRemoteFunction("{class_name}.{method_name}",'
+                    "null,"
+                    f'{{{",".join(
+                        f'"{para}":{para}' for para
+                        in inspect.signature(method).parameters
+                    )}}}'
+                    ").result};"
+                    "\n"
+                )
+    return result
+
+
+async def _inject_javascript(
+    target: Page | BrowserContext, injector: Injector, port: int
+) -> None:
+    with open(
+        Path(__file__).resolve().parent
+        / "template"
+        / "invokeRemoteFunction.js",
+        encoding="utf-8",
+    ) as file:
+        await target.add_init_script(
+            Environment().from_string(file.read()).render(port=port)
+        )
+    await target.add_init_script(_generate_script(injector))
+
+
+async def _override_cors_restrictions(
+    target: Page | BrowserContext, port: int
+) -> None:
+    async def override_cors(route, request):
+        url = request.url
+        if f"localhost:{port}" in url or f"127.0.0.1:{port}" in url:
+            await route.continue_()
+        else:
+            response = await target.request.fetch(request)
+            headers = response.headers
+            headers["Access-Control-Allow-Origin"] = "*"
+            await route.fulfill(response=response, headers=headers)
+
+    await target.route("**/*", override_cors)
+
+
+async def inject(target: Page | BrowserContext, injector: Injector) -> None:
+    port = _get_free_port()
+    server = uvicorn.Server(
+        uvicorn.Config(
+            RemoteInvokeServer(injector=injector).app,
+            port=port,
+            log_level="warning",
+        )
+    )
+
+    async def on_close():
+        await server.shutdown()
+
+    asyncio.create_task(server.serve())
+    await _override_cors_restrictions(target, port)
+    await _inject_javascript(target, injector, port)
+
+    target.once("close", on_close)

src/browser/remote_injector/injector.py

@@ -0,0 +1,52 @@
+import inspect
+from typing import Callable, Iterator
+
+
+class Injector:
+    _classes: dict[str, dict[str, Callable]]
+    _objects: dict[str, Callable]
+
+    def __init__(self) -> None:
+        self._classes = {"_": {}}
+        self._objects = {}
+
+    def __getitem__(self, index: str) -> Callable:
+        return self._objects[index]
+
+    def __len__(self) -> int:
+        return len(self._classes)
+
+    def __iter__(self) -> Iterator[Callable]:
+        return iter(self._classes.items())
+
+    def __contains__(self, item: str) -> bool:
+        return item in self._objects
+
+    def __str__(self) -> str:
+        return str(dict(self._classes))
+
+    def _flatten(
+        self, classes: dict, parent_key: str = ""
+    ) -> dict[str, Callable]:
+        items = []
+        for key, value in classes.items():
+            new_key = f"{parent_key}.{key}" if parent_key else key
+            if isinstance(value, dict):
+                items.extend(self._flatten(value, new_key).items())
+            else:
+                items.append((new_key, value))
+        return dict(items)
+
+    def register(self, obj: Callable) -> Callable:
+        if inspect.isclass(obj):
+            self._classes.setdefault(obj.__name__, {})
+            for name, member in obj.__dict__.items():
+                if isinstance(member, staticmethod):
+                    self._classes[obj.__name__][name] = member
+                    self._objects[f"{obj.__name__}.{name}"] = self._classes[
+                        obj.__name__
+                    ][name]
+        elif inspect.isfunction(obj):
+            self._classes["_"][obj.__name__] = obj
+            self._objects[obj.__name__] = self._classes["_"][obj.__name__]
+        return obj