Fix site checks: 2 fixed, 1 new added + proton activation method (#2869)

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

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"
         },

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