Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
162 changes: 139 additions & 23 deletions aimlib/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@

import asyncio
import importlib
import math
import os
import re
from typing import Optional
Expand Down Expand Up @@ -194,7 +195,10 @@ def _with_google_chrome_brand(rows: list[dict[str, str]], *, label: str) -> list
return branded


def _google_chrome_user_agent_override(identity: object) -> Optional[dict]:
def _google_chrome_user_agent_override(
identity: object,
model_override: Optional[str] = None,
) -> Optional[dict]:
"""Build a coherent ChromePublic to Google Chrome UA-CH override."""
if not isinstance(identity, dict):
raise BrowserPolicyError("managed browser identity metadata is unavailable")
Expand All @@ -216,6 +220,10 @@ def _google_chrome_user_agent_override(identity: object) -> Optional[dict]:
raise BrowserPolicyError("managed browser did not present a mobile Android user agent")
if ua_platform != "Android" or identity.get("mobile") is not True:
raise BrowserPolicyError("managed browser did not present Android client hints")
if model_override is not None and (
not isinstance(model_override, str) or not 1 <= len(model_override) <= 128
):
raise BrowserPolicyError("managed browser footprint model is invalid")

brands = _validated_ua_brand_rows(identity.get("brands"), label="brand")
full_versions = _validated_ua_brand_rows(
Expand All @@ -226,7 +234,8 @@ def _google_chrome_user_agent_override(identity: object) -> Optional[dict]:
full_has_google = any(row["brand"] == "Google Chrome" for row in full_versions)
if low_has_google != full_has_google:
raise BrowserPolicyError("managed browser Chrome brand metadata is inconsistent")
if low_has_google:
effective_model = model_override if model_override is not None else identity["model"]
if low_has_google and effective_model == identity["model"]:
return None

return {
Expand All @@ -241,14 +250,84 @@ def _google_chrome_user_agent_override(identity: object) -> Optional[dict]:
"platform": ua_platform,
"platformVersion": identity["platformVersion"],
"architecture": identity["architecture"],
"model": identity["model"],
"model": effective_model,
"mobile": True,
"bitness": identity["bitness"],
"wow64": identity.get("wow64") is True,
},
}


def _validated_footprint_identity(value: object, applied_footprint: str) -> Optional[dict]:
"""Validate the public, browser-observable identity for an applied footprint."""
if not applied_footprint:
return None
if not isinstance(applied_footprint, str) or len(applied_footprint) > 128:
raise BrowserPolicyError("managed browser footprint identifier is invalid")
if not isinstance(value, dict) or value.get("name") != applied_footprint:
raise BrowserPolicyError("managed browser footprint identity is unavailable")
model = value.get("model")
width = value.get("screen_width")
height = value.get("screen_height")
ratio = value.get("device_pixel_ratio")
if not isinstance(model, str) or not 1 <= len(model) <= 128:
raise BrowserPolicyError("managed browser footprint model is invalid")
if (
isinstance(width, bool)
or not isinstance(width, int)
or not 320 <= width <= 10_000
or isinstance(height, bool)
or not isinstance(height, int)
or not 320 <= height <= 10_000
):
raise BrowserPolicyError("managed browser footprint display is invalid")
if (
isinstance(ratio, bool)
or not isinstance(ratio, (int, float))
or not math.isfinite(float(ratio))
or not 0.5 <= float(ratio) <= 10
):
raise BrowserPolicyError("managed browser footprint pixel ratio is invalid")
return {
"name": applied_footprint,
"model": model,
"screen_width": width,
"screen_height": height,
"device_pixel_ratio": float(ratio),
}


def _device_metrics_override(identity: Optional[dict]) -> Optional[dict]:
"""Convert physical display pixels into the target's portrait CSS viewport."""
if identity is None:
return None
ratio = identity["device_pixel_ratio"]
css_width = math.ceil(identity["screen_width"] / ratio)
css_height = math.ceil(identity["screen_height"] / ratio)
return {
"width": css_width,
"height": css_height,
"deviceScaleFactor": ratio,
"mobile": True,
"screenWidth": css_width,
"screenHeight": css_height,
"positionX": 0,
"positionY": 0,
"screenOrientation": {"type": "portraitPrimary", "angle": 0},
}


def _identity_verification(identity: Optional[dict], metrics: Optional[dict]) -> Optional[dict]:
if identity is None or metrics is None:
return None
return {
"model": identity["model"],
"screenWidth": metrics["screenWidth"],
"screenHeight": metrics["screenHeight"],
"devicePixelRatio": metrics["deviceScaleFactor"],
}


_READ_NATIVE_BROWSER_IDENTITY = r"""async () => {
const data = navigator.userAgentData;
if (!data) return null;
Expand Down Expand Up @@ -279,20 +358,25 @@ def _google_chrome_user_agent_override(identity: object) -> Optional[dict]:
}"""


_VERIFY_GOOGLE_CHROME_BRAND = r"""async () => {
_VERIFY_MANAGED_BROWSER_IDENTITY = r"""async expected => {
const data = navigator.userAgentData;
// UA-CH is unavailable on Chrome's internal startup page. That is not an override failure:
// the same target exposes the metadata after its first secure navigation.
if (!data || !Array.isArray(data.brands)) return null;
let high = {};
try {
high = await data.getHighEntropyValues(['fullVersionList']);
high = await data.getHighEntropyValues(['fullVersionList', 'model']);
} catch (_) {
return null;
}
if (!Array.isArray(high.fullVersionList)) return null;
return data.brands.some(row => row.brand === 'Google Chrome') &&
const branded = data.brands.some(row => row.brand === 'Google Chrome') &&
high.fullVersionList.some(row => row.brand === 'Google Chrome');
if (!branded || !expected) return branded;
return high.model === expected.model &&
screen.width === expected.screenWidth &&
screen.height === expected.screenHeight &&
Math.abs(devicePixelRatio - expected.devicePixelRatio) < 0.001;
}"""


Expand Down Expand Up @@ -485,7 +569,7 @@ def __init__(self, ai: "Aimlib", device: Device, data: dict):
self.max_tabs = int(data.get("max_tabs") or 5) # per-session tab cap (server-configured)
self._gave_initial = False
self.egress_ip = None
self.fingerprint = {}
self.fingerprint = data.get("resolved_profile") or {}
self.desired_footprint = data.get("desired_footprint") or ""
self.applied_footprint = data.get("applied_footprint") or ""
self.expires_at = data.get("expires_at")
Expand Down Expand Up @@ -516,6 +600,7 @@ async def wait_until_ready(self, timeout: float = 180):
# Do not connect until a requested browser footprint is fully active.
footprint_ok = (not self.desired_footprint) or (self.applied_footprint == self.desired_footprint)
if self.status in ("ready", "active", "idle") and footprint_ok:
_validated_footprint_identity(self.fingerprint, self.applied_footprint)
return
if self.status in ("expiring", "gone"):
raise SessionExpiredError(f"session {self.status}")
Expand Down Expand Up @@ -611,9 +696,15 @@ async def set_footprint(self, footprint: Optional[str], timeout: float = 120):
r = await self._ai._http.get(f"/v1/devices/{self.device.id}/browser")
if r.status_code == 404:
raise SessionExpiredError("session not found / expired")
_raise_for_typed(r)
data = r.json()
if data.get("session_id") != self.id:
raise SessionExpiredError("session was replaced by a newer browser session")
self.status = data.get("status")
self.applied_footprint = data.get("applied_footprint") or ""
if self.applied_footprint == fp and data.get("status") in ("ready", "active", "idle"):
self.fingerprint = data.get("resolved_profile") or {}
if self.applied_footprint == fp and self.status in ("ready", "active", "idle"):
_validated_footprint_identity(self.fingerprint, self.applied_footprint)
await self.connect(timeout=timeout) # transparent reconnect to the same session_id
return
await asyncio.sleep(2)
Expand Down Expand Up @@ -646,17 +737,31 @@ async def _apply_page_identity(self, page):
return
cdp_session = None
try:
identity = await self._native_identity_for_page(page)
override = _google_chrome_user_agent_override(identity)
if override is not None:
footprint_identity = _validated_footprint_identity(
self.fingerprint,
self.applied_footprint,
)
identity = await self._native_identity_for_page(page, footprint_identity)
override = _google_chrome_user_agent_override(
identity,
footprint_identity["model"] if footprint_identity else None,
)
metrics = _device_metrics_override(footprint_identity)
if override is not None or metrics is not None:
cdp_session = await page.context.new_cdp_session(page)
await cdp_session.send("Emulation.setUserAgentOverride", override)
verified = await page.evaluate(_VERIFY_GOOGLE_CHROME_BRAND)
if override is not None:
await cdp_session.send("Emulation.setUserAgentOverride", override)
if metrics is not None:
await cdp_session.send("Emulation.setDeviceMetricsOverride", metrics)
verified = await page.evaluate(
_VERIFY_MANAGED_BROWSER_IDENTITY,
_identity_verification(footprint_identity, metrics),
)
# Internal/about:blank startup pages cannot expose UA-CH. The override has already
# been validated on the secure bootstrap tab below; a definitive false result on a
# page that does expose UA-CH still fails closed.
if verified is False:
raise BrowserPolicyError("managed browser Chrome branding did not apply")
raise BrowserPolicyError("managed browser identity did not apply")
self._identity_cdp_sessions.append(cdp_session)
cdp_session = None
self._identity_pages.add(page_key)
Expand All @@ -671,7 +776,7 @@ async def _apply_page_identity(self, page):
except Exception: # noqa: BLE001
pass

async def _native_identity_for_page(self, page):
async def _native_identity_for_page(self, page, footprint_identity: Optional[dict] = None):
"""Read UA-CH in a secure first-party context when Chrome's startup tab cannot expose it."""
if self._native_browser_identity is not None:
return self._native_browser_identity
Expand All @@ -687,21 +792,32 @@ async def _native_identity_for_page(self, page):
timeout=30_000,
)
identity = await bootstrap_page.evaluate(_READ_NATIVE_BROWSER_IDENTITY)
override = _google_chrome_user_agent_override(identity)
if override is not None:
override = _google_chrome_user_agent_override(
identity,
footprint_identity["model"] if footprint_identity else None,
)
metrics = _device_metrics_override(footprint_identity)
if override is not None or metrics is not None:
bootstrap_cdp_session = await page.context.new_cdp_session(
bootstrap_page
)
await bootstrap_cdp_session.send(
"Emulation.setUserAgentOverride",
override,
)
if override is not None:
await bootstrap_cdp_session.send(
"Emulation.setUserAgentOverride",
override,
)
if metrics is not None:
await bootstrap_cdp_session.send(
"Emulation.setDeviceMetricsOverride",
metrics,
)
verified = await bootstrap_page.evaluate(
_VERIFY_GOOGLE_CHROME_BRAND
_VERIFY_MANAGED_BROWSER_IDENTITY,
_identity_verification(footprint_identity, metrics),
)
if verified is not True:
raise BrowserPolicyError(
"managed browser Chrome branding did not apply"
"managed browser identity did not apply"
)
except Exception as exc: # noqa: BLE001 - keep URL/session details out of the error
if isinstance(exc, BrowserPolicyError):
Expand Down
2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
[project]
name = "aimlib"
version = "0.4.2"
version = "0.4.3"
description = "Python SDK for aimlib mobile proxies and remote-browser sessions"
readme = "README.md"
requires-python = ">=3.10"
Expand Down
Loading