From 74c8bb77a5754a26b5763c51e3f124477bb9d8e6 Mon Sep 17 00:00:00 2001 From: Anthony Volk Date: Fri, 15 May 2026 19:52:11 +0200 Subject: [PATCH 1/2] Handle PyPI login confirmation flow --- src/main/python/pypi_cleanup/__init__.py | 159 +++++++++++++++++++--- src/unittest/python/pypi_cleanup_tests.py | 110 +++++++++++++++ 2 files changed, 250 insertions(+), 19 deletions(-) diff --git a/src/main/python/pypi_cleanup/__init__.py b/src/main/python/pypi_cleanup/__init__.py index 72b7028..9046922 100644 --- a/src/main/python/pypi_cleanup/__init__.py +++ b/src/main/python/pypi_cleanup/__init__.py @@ -36,7 +36,30 @@ DEFAULT_PATTERNS = [re.compile(r".*\.dev\d+$")] -class CsfrParser(HTMLParser): +class PageTitleParser(HTMLParser): + def __init__(self): + super().__init__() + self._in_title = False + self._title = [] + + def handle_starttag(self, tag, attrs): + if tag == "title": + self._in_title = True + + def handle_data(self, data): + if self._in_title: + self._title.append(data.strip()) + + def handle_endtag(self, tag): + if tag == "title": + self._in_title = False + + @property + def title(self): + return " ".join(part for part in self._title if part) + + +class CsrfParser(HTMLParser): def __init__(self, target, contains_input=None): super().__init__() self._target = target @@ -50,7 +73,9 @@ def handle_starttag(self, tag, attrs): if tag == "form": attrs = dict(attrs) action = attrs.get("action") # Might be None. - if action and (action == self._target or action.startswith(self._target)): + self._csrf = None + self._input_contained = False + if action and self._form_action_matches(action): self._in_form = True return @@ -64,6 +89,13 @@ def handle_starttag(self, tag, attrs): return + def _form_action_matches(self, action): + if action == self._target or action.startswith(self._target): + return True + + parsed = urlparse(action) + return bool(parsed.path and (parsed.path == self._target or parsed.path.startswith(self._target))) + def handle_endtag(self, tag): if tag == "form": self._in_form = False @@ -73,6 +105,9 @@ def handle_endtag(self, tag): return +CsfrParser = CsrfParser + + class PypiCleanup: def __init__(self, url, username, packages, do_it, patterns, verbose, days, query_only, leave_most_recent_only, confirm, delete_project, **_): @@ -90,6 +125,99 @@ def __init__(self, url, username, packages, do_it, patterns, verbose, days, quer self.leave_most_recent_only = leave_most_recent_only self.date = datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(days=days) + def _relative_url(self, url): + if url.startswith(self.url): + return url[len(self.url):] + return url + + def _redact_url(self, url): + parsed = urlparse(url) + if parsed.query: + return parsed._replace(query="").geturl() + return url + + def _is_login_url(self, url): + return self._relative_url(url).startswith("/account/login/") + + def _is_two_factor_url(self, url): + return self._relative_url(url).startswith("/account/two-factor/") + + def _is_confirm_login_url(self, url): + return self._relative_url(url).startswith("/account/confirm-login/") + + def _is_reauthenticate_url(self, url): + return self._relative_url(url).startswith("/account/reauthenticate/") + + def _page_title(self, html): + parser = PageTitleParser() + parser.feed(html) + return parser.title + + def _missing_csrf_message(self, response, form_action, contains_input=None): + url = getattr(response, "url", "") + title = self._page_title(getattr(response, "text", "")) + expected = f" form containing {contains_input!r}" if contains_input else " form" + message = f"No CSRF token found in expected{expected} for {form_action}; final URL: {self._redact_url(url)}" + if title: + message = f"{message}; page title: {title!r}" + + if self._is_confirm_login_url(url): + message = ( + f"{message}. PyPI requires email confirmation for this login. " + "Confirm the login from the PyPI email and rerun the command, or paste the confirmation URL when prompted." + ) + elif self._is_login_url(url): + message = f"{message}. PyPI returned the login page, so authentication did not complete." + elif self._is_reauthenticate_url(url): + message = f"{message}. PyPI requires password reauthentication before this action." + + return message + + def _csrf_from_response(self, response, form_action, contains_input=None): + parser = CsrfParser(form_action, contains_input) + parser.feed(response.text) + if not parser.csrf: + raise ValueError(self._missing_csrf_message(response, form_action, contains_input)) + return parser.csrf + + def _complete_email_login_confirmation(self, session): + logging.warning("PyPI requires email confirmation for this login.") + logging.warning("Open the PyPI email and copy the full confirmation URL.") + confirmation_url = getpass.getpass( + "Paste PyPI login confirmation URL (input hidden, press Enter to abort): " + ).strip() + + if not confirmation_url: + logging.error("Login confirmation was not provided") + return False + + parsed = urlparse(confirmation_url) + expected = urlparse(self.url) + if ( + parsed.scheme != expected.scheme + or parsed.netloc != expected.netloc + or parsed.path != "/account/confirm-login/" + or "token=" not in parsed.query + ): + logging.error("Refusing unexpected PyPI login confirmation URL") + return False + + with session.get(confirmation_url, headers={"referer": f"{self.url}/account/confirm-login/"}) as r: + r.raise_for_status() + if self._is_login_url(r.url) or self._is_confirm_login_url(r.url): + logging.error(f"PyPI did not accept the login confirmation URL; final URL: {self._redact_url(r.url)}") + return False + + return self._verify_logged_in(session) + + def _verify_logged_in(self, session): + with session.get(f"{self.url}/manage/projects/") as r: + r.raise_for_status() + if self._is_login_url(r.url) or self._is_confirm_login_url(r.url): + logging.error(f"PyPI login did not complete; final URL: {self._redact_url(r.url)}") + return False + return True + def run(self): csrf = None @@ -223,11 +351,7 @@ def package_matches_file(p, v, f): with s.get(f"{self.url}/account/login/") as r: r.raise_for_status() form_action = "/account/login/" - parser = CsfrParser(form_action) - parser.feed(r.text) - if not parser.csrf: - raise ValueError(f"No CSFR found in {form_action}") - csrf = parser.csrf + csrf = self._csrf_from_response(r, form_action) two_factor = False with s.post(f"{self.url}/account/login/", @@ -236,17 +360,13 @@ def package_matches_file(p, v, f): "password": password}, headers={"referer": f"{self.url}/account/login/"}) as r: r.raise_for_status() - if r.url == f"{self.url}/account/login/": + if self._is_login_url(r.url): logging.error(f"Login for user {self.username} failed") return 1 - if r.url.startswith(f"{self.url}/account/two-factor/"): + if self._is_two_factor_url(r.url): form_action = r.url[len(self.url):] - parser = CsfrParser(form_action) - parser.feed(r.text) - if not parser.csrf: - raise ValueError(f"No CSFR found in {form_action}") - csrf = parser.csrf + csrf = self._csrf_from_response(r, form_action) two_factor = True two_factor_url = r.url @@ -260,6 +380,11 @@ def package_matches_file(p, v, f): if r.url == two_factor_url: logging.error(f"Authentication code {auth_code} is invalid") return 1 + if self._is_confirm_login_url(r.url) and not self._complete_email_login_confirmation(s): + return 1 + + if not self._verify_logged_in(s): + return 1 if self.do_it: logging.warning("!!! WILL ACTUALLY DELETE THINGS - LAST CHANCE TO CHANGE YOUR MIND !!!") @@ -274,11 +399,7 @@ def package_matches_file(p, v, f): form_url = f"{self.url}{form_action}" with s.get(form_url) as r: r.raise_for_status() - parser = CsfrParser(form_action, "confirm_delete_version") - parser.feed(r.text) - if not parser.csrf: - raise ValueError(f"No CSFR found in {form_action}") - csrf = parser.csrf + csrf = self._csrf_from_response(r, form_action, "confirm_delete_version") referer = r.url with s.post(form_url, diff --git a/src/unittest/python/pypi_cleanup_tests.py b/src/unittest/python/pypi_cleanup_tests.py index cbf5ba4..0dcbd3b 100644 --- a/src/unittest/python/pypi_cleanup_tests.py +++ b/src/unittest/python/pypi_cleanup_tests.py @@ -19,6 +19,50 @@ from pypi_cleanup import PypiCleanup +class ContextResponse: + def __init__(self, url, text=""): + self.url = url + self.text = text + + def __enter__(self): + return self + + def __exit__(self, *args): + return False + + def raise_for_status(self): + return None + + +class FakeConfirmationSession: + def __init__(self): + self.urls = [] + + def get(self, url, headers=None): + self.urls.append((url, headers)) + if url.startswith("https://pypi.org/account/confirm-login/"): + return ContextResponse("https://pypi.org/manage/projects/") + if url == "https://pypi.org/manage/projects/": + return ContextResponse("https://pypi.org/manage/projects/") + raise AssertionError(f"Unexpected URL {url}") + + +def cleanup_for_tests(): + return PypiCleanup( + url="https://pypi.org", + username="user", + packages=["test-package"], + do_it=False, + patterns=None, + verbose=False, + days=0, + query_only=False, + leave_most_recent_only=False, + confirm=False, + delete_project=False + ) + + class TestEmptyMatchesListRegression(unittest.TestCase): """ Regression test for the bug where max() is called on an empty list when @@ -82,5 +126,71 @@ def test_version_with_no_matching_files_does_not_crash(self, mock_session): raise +class TestCsrfParsing(unittest.TestCase): + def test_delete_release_form_csrf_is_found(self): + cleanup = cleanup_for_tests() + response = ContextResponse( + "https://pypi.org/manage/project/test-package/release/1.0.0/", + """ +
+ + +
+ """, + ) + + self.assertEqual( + cleanup._csrf_from_response( + response, + "/manage/project/test-package/release/1.0.0/", + "confirm_delete_version", + ), + "csrf-value", + ) + + def test_missing_delete_form_explains_email_confirmation(self): + cleanup = cleanup_for_tests() + response = ContextResponse( + "https://pypi.org/account/confirm-login/", + "Please confirm this login", + ) + + with self.assertRaisesRegex(ValueError, "PyPI requires email confirmation"): + cleanup._csrf_from_response( + response, + "/manage/project/test-package/release/1.0.0/", + "confirm_delete_version", + ) + + +class TestLoginConfirmation(unittest.TestCase): + @patch("pypi_cleanup.getpass.getpass") + def test_email_confirmation_url_is_followed_in_same_session(self, mock_getpass): + cleanup = cleanup_for_tests() + session = FakeConfirmationSession() + mock_getpass.return_value = "https://pypi.org/account/confirm-login/?token=abc123" + + self.assertTrue(cleanup._complete_email_login_confirmation(session)) + self.assertEqual( + session.urls, + [ + ( + "https://pypi.org/account/confirm-login/?token=abc123", + {"referer": "https://pypi.org/account/confirm-login/"}, + ), + ("https://pypi.org/manage/projects/", None), + ], + ) + + @patch("pypi_cleanup.getpass.getpass") + def test_unexpected_confirmation_url_is_rejected(self, mock_getpass): + cleanup = cleanup_for_tests() + session = FakeConfirmationSession() + mock_getpass.return_value = "https://example.com/account/confirm-login/?token=abc123" + + self.assertFalse(cleanup._complete_email_login_confirmation(session)) + self.assertEqual(session.urls, []) + + if __name__ == '__main__': unittest.main() From c533688662dc1e0dc360c91faabe36b72b17e3d1 Mon Sep 17 00:00:00 2001 From: Anthony Volk Date: Fri, 15 May 2026 20:47:58 +0200 Subject: [PATCH 2/2] Document PyPI login confirmation flow --- README.md | 26 ++++++++++++- src/main/python/pypi_cleanup/__init__.py | 47 +++++++++++++++++++++++- 2 files changed, 71 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index dabe022..82b6a46 100644 --- a/README.md +++ b/README.md @@ -34,11 +34,23 @@ Authentication password may be passed via environment variable Authentication with TOTP is supported. +PyPI may require an additional email confirmation step for TOTP logins from a new or unrecognized device. +When PyPI redirects to the login confirmation page, `pypi-cleanup` prompts for the confirmation URL from the +email and follows it with the same authenticated session before continuing. This supports PyPI's newer +unrecognized-device login verification flow; see PyPI's note on +[login verification protections](https://blog.pypi.org/posts/2025-11-14-login-verification/) and the related +`pypi-cleanup` report about missing release-page CSRF tokens in +[issue #42](https://github.com/arcivanov/pypi-cleanup/issues/42). + +If authentication does not complete, `--debug-auth` logs the final URLs, page titles, and high-level page markers +for the login flow. It also writes sanitized HTML snapshots with CSRF tokens, TOTP values, passwords, and URL tokens +redacted. + ### Examples: ```bash $ pypi-cleanup --help -usage: pypi-cleanup [-h] [-u USERNAME] -p PACKAGES [-t URL] [-r PATTERNS | --leave-most-recent-only] [--query-only] [--do-it] [--delete-project] [-y] [-d DAYS] [-v] +usage: pypi-cleanup [-h] [-u USERNAME] -p PACKAGES [-t URL] [-r PATTERNS | --leave-most-recent-only] [--query-only] [--do-it] [--delete-project] [-y] [-d DAYS] [-v] [--debug-auth] PyPi Package Cleanup Utility v0.1.8 @@ -59,6 +71,7 @@ options: -y, --yes confirm extremely dangerous destructive delete (default: False) -d DAYS, --days DAYS only delete releases **matching specified patterns** where all files are older than X days (default: 0) -v, --verbose be verbose (default: 0) + --debug-auth log PyPI authentication redirects, page titles, markers, and sanitized snapshots (default: False) ``` #### Query-Only Mode @@ -105,6 +118,17 @@ INFO:root:Would be deleting 'pybuilder' version 0.13.13.dev20240604074936, but n INFO:root:Would be deleting 'pybuilder' version 0.13.14.dev20240814015648, but not doing it! ``` +If PyPI requires email confirmation for the login, the command pauses after TOTP: + +```bash +Password: +Authentication code: 123456 +WARNING:root:PyPI requires email confirmation for this login. +WARNING:root:Open the PyPI email and copy the full confirmation URL. +Paste PyPI login confirmation URL (input hidden, press Enter to abort): +INFO:root:Would be deleting 'pybuilder' version 0.13.13.dev20240604074936, but not doing it! +``` + Now to actually delete the specificed packages ```bash $ pypi-cleanup -u arcivanov -p pybuilder --do-it diff --git a/src/main/python/pypi_cleanup/__init__.py b/src/main/python/pypi_cleanup/__init__.py index 9046922..f9abd96 100644 --- a/src/main/python/pypi_cleanup/__init__.py +++ b/src/main/python/pypi_cleanup/__init__.py @@ -22,6 +22,7 @@ import os import re import sys +import tempfile import time from html.parser import HTMLParser from textwrap import dedent @@ -110,7 +111,7 @@ def handle_endtag(self, tag): class PypiCleanup: def __init__(self, url, username, packages, do_it, patterns, verbose, days, query_only, leave_most_recent_only, - confirm, delete_project, **_): + confirm, delete_project, debug_auth=False, **_): self.url = urlparse(url).geturl() if self.url[-1] == "/": self.url = self.url[:-1] @@ -123,6 +124,7 @@ def __init__(self, url, username, packages, do_it, patterns, verbose, days, quer self.verbose = verbose self.query_only = query_only self.leave_most_recent_only = leave_most_recent_only + self.debug_auth = debug_auth self.date = datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(days=days) def _relative_url(self, url): @@ -153,6 +155,44 @@ def _page_title(self, html): parser.feed(html) return parser.title + def _auth_page_markers(self, html): + lower = html.lower() + markers = ("unrecognized", "confirm", "email", "login", "two-factor", "reauthenticate") + return [marker for marker in markers if marker in lower] + + def _sanitize_html(self, html): + html = re.sub(r'(?i)(name="csrf_token"[^>]*value=")[^"]*"', r'\1"', html) + html = re.sub(r'(?i)(name="totp_value"[^>]*value=")[^"]*"', r'\1"', html) + html = re.sub(r'(?i)(name="password"[^>]*value=")[^"]*"', r'\1"', html) + html = re.sub(r'(?i)(token=)[^"&\s<]+', r'\1', html) + return html + + def _write_auth_snapshot(self, label, response): + with tempfile.NamedTemporaryFile( + "w", + encoding="utf-8", + delete=False, + prefix=f"pypi-cleanup-{label}-", + suffix=".html", + ) as f: + f.write(self._sanitize_html(response.text)) + return f.name + + def _debug_auth_response(self, label, response, snapshot=False): + if not self.debug_auth: + return + + logging.info( + "Auth debug [%s]: status=%s final_url=%s title=%r markers=%s", + label, + getattr(response, "status_code", ""), + self._redact_url(getattr(response, "url", "")), + self._page_title(getattr(response, "text", "")), + self._auth_page_markers(getattr(response, "text", "")), + ) + if snapshot: + logging.info("Auth debug [%s]: sanitized snapshot=%s", label, self._write_auth_snapshot(label, response)) + def _missing_csrf_message(self, response, form_action, contains_input=None): url = getattr(response, "url", "") title = self._page_title(getattr(response, "text", "")) @@ -350,6 +390,7 @@ def package_matches_file(p, v, f): with s.get(f"{self.url}/account/login/") as r: r.raise_for_status() + self._debug_auth_response("login-get", r) form_action = "/account/login/" csrf = self._csrf_from_response(r, form_action) @@ -360,6 +401,7 @@ def package_matches_file(p, v, f): "password": password}, headers={"referer": f"{self.url}/account/login/"}) as r: r.raise_for_status() + self._debug_auth_response("login-post", r, snapshot=True) if self._is_login_url(r.url): logging.error(f"Login for user {self.username} failed") return 1 @@ -377,6 +419,7 @@ def package_matches_file(p, v, f): "totp_value": auth_code}, headers={"referer": two_factor_url}) as r: r.raise_for_status() + self._debug_auth_response("totp-post", r, snapshot=True) if r.url == two_factor_url: logging.error(f"Authentication code {auth_code} is invalid") return 1 @@ -443,6 +486,8 @@ def main(): help="only delete releases **matching specified patterns** where all files are " "older than X days") parser.add_argument("-v", "--verbose", action="store_const", const=1, default=0, help="be verbose") + parser.add_argument("--debug-auth", action="store_true", default=False, + help="log PyPI authentication redirects, page titles, markers, and sanitized snapshots") args = parser.parse_args() if args.patterns and not args.confirm and not args.do_it and not args.query_only: