diff --git a/maigret/activation.py b/maigret/activation.py index a44ee8e..12285dd 100644 --- a/maigret/activation.py +++ b/maigret/activation.py @@ -122,6 +122,33 @@ class ParsingActivator: site.headers["sign"] = sg logger.debug(f"OnlyFans signed {target_path} time={t}") + @staticmethod + async def proton(site, logger, **kwargs): + # Proton's /api/users/available now requires an anon session: POST + # /api/auth/v4/sessions returns UID + AccessToken which must be sent + # as x-pm-uid and Authorization: Bearer on the availability call. + headers = { + k: v for k, v in site.headers.items() + if k.lower() not in ("authorization", "x-pm-uid") + } + async with ClientSession(trust_env=True) as session: + async with session.post( + site.activation["url"], + headers=headers, + json={}, + timeout=kwargs.get("timeout"), + ) as response: + payload = await response.json(content_type=None) + uid, token = payload.get("UID"), payload.get("AccessToken") + if uid and token: + site.headers["x-pm-uid"] = uid + site.headers["Authorization"] = f"Bearer {token}" + logger.debug("Proton activation: got session UID + token") + else: + logger.warning( + f"Proton activation failed: no UID/token in {str(payload)[:120]!r}" + ) + @staticmethod async def weibo(site, logger, **kwargs): # Weibo gates its ajax profile API behind an anonymous "Sina Visitor diff --git a/maigret/resources/data.json b/maigret/resources/data.json index ffd482d..886616c 100644 --- a/maigret/resources/data.json +++ b/maigret/resources/data.json @@ -6544,6 +6544,24 @@ "challenges.cloudflare.com": "Cloudflare challenge" } }, + "Amateur.tv": { + "protection": [ + "tls_fingerprint" + ], + "tags": [ + "porn", + "webcam" + ], + "checkType": "status_code", + "urlMain": "https://amateur.tv/", + "url": "https://amateur.tv/{username}", + "usernameClaimed": "blue", + "usernameUnclaimed": "noooonewouldeverusethis7abcdef", + "errors": { + "Just a moment": "Cloudflare challenge", + "challenges.cloudflare.com": "Cloudflare challenge" + } + }, "Techdirt": { "disabled": true, "tags": [ @@ -10288,6 +10306,9 @@ "forum" ], "engine": "vBulletin", + "protection": [ + "tls_fingerprint" + ], "urlMain": "https://forum.blu-ray.com/", "usernameClaimed": "adam", "usernameUnclaimed": "noonewouldeverusethis7" @@ -27300,12 +27321,18 @@ "Username already used" ], "absenceStrs": [ - "\"Code\": 1000" + "\"Code\":1000" ], "headers": { - "X-Pm-Appversion": "web-account@4.28.2" + "X-Pm-Appversion": "web-account@5.0.398.1" }, - "url": "https://account.protonmail.com/api/users/available?Name={username}", + "activation": { + "url": "https://account.proton.me/api/auth/v4/sessions", + "method": "proton", + "marks": ["Invalid access token"] + }, + "urlMain": "https://proton.me/", + "url": "https://account.proton.me/api/users/available?Name={username}", "usernameClaimed": "adam", "usernameUnclaimed": "noonewouldeverusethis7" }, diff --git a/tests/test_activation.py b/tests/test_activation.py index 49c5388..0483edb 100644 --- a/tests/test_activation.py +++ b/tests/test_activation.py @@ -94,7 +94,7 @@ class _FakeResponse: return self._json_data -@pytest.mark.parametrize("method", ["twitter", "vimeo", "onlyfans", "weibo"]) +@pytest.mark.parametrize("method", ["twitter", "vimeo", "onlyfans", "weibo", "proton"]) def test_activation_methods_are_coroutines(method): assert inspect.iscoroutinefunction(getattr(ParsingActivator, method)) @@ -260,6 +260,112 @@ async def test_onlyfans_sign_differs_per_path(monkeypatch): assert sig_adam != sig_bob +@pytest.mark.asyncio +async def test_proton_activation_sets_uid_and_bearer(monkeypatch): + """Proton activator bootstraps an anon session and injects UID + Bearer token.""" + site = _FakeSite( + headers={"X-Pm-Appversion": "web-account@5.0.398.1"}, + activation={"url": "https://account.proton.me/api/auth/v4/sessions"}, + ) + captured = {} + + class FakeSession: + def __init__(self, **kwargs): + captured["session_kwargs"] = kwargs + + async def __aenter__(self): + return self + + async def __aexit__(self, exc_type, exc, tb): + return False + + def post(self, url, headers=None, json=None, timeout=None): + captured["url"] = url + captured["headers"] = dict(headers or {}) + captured["json"] = json + captured["timeout"] = timeout + return _FakeResponse(json_data={"UID": "uid123", "AccessToken": "tok456"}) + + monkeypatch.setattr("maigret.activation.ClientSession", FakeSession) + + await ParsingActivator.proton(site, Mock(), timeout=7) + + assert captured["url"] == "https://account.proton.me/api/auth/v4/sessions" + assert captured["headers"] == {"X-Pm-Appversion": "web-account@5.0.398.1"} + assert captured["json"] == {} + assert captured["timeout"] == 7 + assert site.headers["x-pm-uid"] == "uid123" + assert site.headers["Authorization"] == "Bearer tok456" + + +@pytest.mark.asyncio +async def test_proton_activation_strips_stale_auth_from_bootstrap(monkeypatch): + """A prior Authorization/x-pm-uid must not be sent to the session endpoint.""" + site = _FakeSite( + headers={ + "X-Pm-Appversion": "web-account@5.0.398.1", + "Authorization": "Bearer stale", + "x-pm-uid": "stale-uid", + }, + activation={"url": "https://account.proton.me/api/auth/v4/sessions"}, + ) + captured = {} + + class FakeSession: + def __init__(self, **kwargs): + pass + + async def __aenter__(self): + return self + + async def __aexit__(self, exc_type, exc, tb): + return False + + def post(self, url, headers=None, json=None, timeout=None): + captured["headers"] = dict(headers or {}) + return _FakeResponse(json_data={"UID": "u", "AccessToken": "t"}) + + monkeypatch.setattr("maigret.activation.ClientSession", FakeSession) + + await ParsingActivator.proton(site, Mock()) + + assert "Authorization" not in captured["headers"] + assert "x-pm-uid" not in captured["headers"] + assert site.headers["Authorization"] == "Bearer t" + assert site.headers["x-pm-uid"] == "u" + + +@pytest.mark.asyncio +async def test_proton_activation_missing_token_leaves_headers_untouched(monkeypatch): + """If Proton returns an error payload, do not inject broken auth headers.""" + site = _FakeSite( + headers={"X-Pm-Appversion": "web-account@5.0.398.1"}, + activation={"url": "https://account.proton.me/api/auth/v4/sessions"}, + ) + + class FakeSession: + def __init__(self, **kwargs): + pass + + async def __aenter__(self): + return self + + async def __aexit__(self, exc_type, exc, tb): + return False + + def post(self, url, headers=None, json=None, timeout=None): + return _FakeResponse(json_data={"Code": 5002, "Error": "Missing header"}) + + monkeypatch.setattr("maigret.activation.ClientSession", FakeSession) + + logger = Mock() + await ParsingActivator.proton(site, logger) + + assert "Authorization" not in site.headers + assert "x-pm-uid" not in site.headers + logger.warning.assert_called_once() + + @pytest.mark.asyncio async def test_wikimapia_activation_parses_token_from_challenge(): """The Wikimapia activator reads the ngxsession token from the challenge