Merge branch 'main' into fix-fromfd-test-race

Author: asvetlov

Lib/_pyrepl/utils.py

@@ -19,7 +19,21 @@
 from .trace import trace
 
 
-ANSI_ESCAPE_SEQUENCE = re.compile(r"\x1b\[[ -@]*[A-~]")
+ANSI_ESCAPE_SEQUENCE = re.compile(
+    r"""
+    \x1b
+    (?:
+        # CSI: ESC [ followed by parameter/intermediate bytes and a final byte.
+        # For example, ESC [ 31 m ("\x1b[31m") selects red foreground text.
+        \[ [ -@]* [A-~]
+        |
+        # OSC: ESC ] followed by a payload terminated by BEL or ST (ESC \).
+        # For example, ESC ] 0 ; title BEL ("\x1b]0;title\x07") sets the title.
+        \] [^\x1b\x07]* (?: \x07 | \x1b\\ )
+    )
+    """,
+    re.X,
+)
 ZERO_WIDTH_BRACKET = re.compile(r"\x01.*?\x02")
 ZERO_WIDTH_TRANS = str.maketrans({"\x01": "", "\x02": ""})
 IDENTIFIERS_AFTER = frozenset({"def", "class"})

Lib/sqlite3/__main__.py

@@ -15,6 +15,13 @@
 from ._completer import completer
 
 
+# IDLE uses CTRL-D for EOF on all platforms (gh-102628).
+if sys.platform == "win32" and "idlelib.run" not in sys.modules:
+    EOF_KEY = "CTRL-Z"
+else:
+    EOF_KEY = "CTRL-D"
+
+
 def execute(c, sql, suppress_errors=True, theme=theme_no_color):
     """Helper that wraps execution of SQL code.
 
@@ -70,7 +77,7 @@ def runsource(self, source, filename="<input>", symbol="single"):
                     print(f"Enter SQL code or one of the below commands, and press enter.\n\n"
                           f"{t.builtin}.version{t.reset}    Print underlying SQLite library version\n"
                           f"{t.builtin}.help{t.reset}       Print this help message\n"
-                          f"{t.builtin}.quit{t.reset}       Exit the CLI, equivalent to CTRL-D\n")
+                          f"{t.builtin}.quit{t.reset}       Exit the CLI, equivalent to {EOF_KEY}\n")
                 case "quit":
                     sys.exit(0)
                 case "":
@@ -115,16 +122,12 @@ def main(*args):
         db_name = repr(args.filename)
 
     # Prepare REPL banner and prompts.
-    if sys.platform == "win32" and "idlelib.run" not in sys.modules:
-        eofkey = "CTRL-Z"
-    else:
-        eofkey = "CTRL-D"
     banner = dedent(f"""
         sqlite3 shell, running on SQLite version {sqlite3.sqlite_version}
         Connected to {db_name}
 
         Each command will be run using execute() on the cursor.
-        Type ".help" for more information; type ".quit" or {eofkey} to quit.
+        Type ".help" for more information; type ".quit" or {EOF_KEY} to quit.
     """).strip()
 
     theme = get_theme()

Lib/test/test_pyrepl/test_render.py

@@ -55,6 +55,15 @@ def test_from_rendered_text_empty_string(self):
         self.assertEqual(line.text, "")
         self.assertEqual(line.width, 0)
 
+    def test_from_rendered_text_with_osc_control(self):
+        osc = "\x1b]633;A\x1b\\"
+        prompt = ">>> "
+        line = RenderLine.from_rendered_text(osc + prompt)
+
+        self.assertEqual(line.width, len(prompt))
+        self.assertEqual(line.cells[0].controls, (osc,))
+        self.assertEqual(line.cells[0].text, "")
+
     def test_from_rendered_text_with_non_sgr_controls(self):
         # \x1b[H is a cursor-home control (not SGR since it doesn't end with 'm')
         line = RenderLine.from_rendered_text("\x1b[Hx")

Lib/test/test_pyrepl/test_utils.py

@@ -47,6 +47,14 @@ def test_wlen(self):
         self.assertEqual(wlen('e\N{COMBINING ACUTE ACCENT}'), 1)
         self.assertEqual(wlen('a\N{ZERO WIDTH JOINER}b'), 2)
 
+    def test_wlen_with_osc_sequence(self):
+        prompt = ">>> "
+        self.assertEqual(wlen("\x1b]633;A\x07" + prompt), len(prompt))
+
+    def test_wlen_with_unterminated_osc_sequence_as_individual_characters(self):
+        # \x1b is treated as a single character (ESC)
+        self.assertEqual(wlen("\x1b]633;A"), len("\x1b]633;A"))
+
     def test_prev_next_window(self):
         def gen_normal():
             yield 1

Lib/test/test_sqlite3/test_cli.py

@@ -82,6 +82,7 @@ class InteractiveSession(unittest.TestCase):
     MEMORY_DB_MSG = "Connected to a transient in-memory database"
     PS1 = "\001\002sqlite> \001\002"
     PS2 = "\001\002    ... \001\002"
+    EOF_KEY = "CTRL-Z" if sys.platform == "win32" else "CTRL-D"
 
     def run_cli(self, *args, commands=()):
         with (
@@ -104,11 +105,19 @@ def run_cli(self, *args, commands=()):
     def test_interact(self):
         out, err = self.run_cli()
         self.assertIn(self.MEMORY_DB_MSG, err)
-        self.assertIn(self.MEMORY_DB_MSG, err)
+        self.assertIn(f'type ".quit" or {self.EOF_KEY} to quit', err)
         self.assertEndsWith(out, self.PS1)
         self.assertEqual(out.count(self.PS1), 1)
         self.assertEqual(out.count(self.PS2), 0)
 
+    def test_interact_help(self):
+        out, err = self.run_cli(commands=(".help",))
+        self.assertIn(self.MEMORY_DB_MSG, err)
+        self.assertIn(f"Exit the CLI, equivalent to {self.EOF_KEY}", out)
+        self.assertEndsWith(out, self.PS1)
+        self.assertEqual(out.count(self.PS1), 2)
+        self.assertEqual(out.count(self.PS2), 0)
+
     def test_interact_quit(self):
         out, err = self.run_cli(commands=(".quit",))
         self.assertIn(self.MEMORY_DB_MSG, err)

Lib/test/test_urllib2.py

@@ -1751,6 +1751,66 @@ def test_proxy_basic_auth(self):
                               "proxy.example.com:3128",
                               )
 
+    def test_proxy_digest_auth(self):
+        # gh-60299: ProxyDigestAuthHandler must look up credentials using
+        # the proxy's authority
+        opener = OpenerDirector()
+        ph = urllib.request.ProxyHandler(dict(http="proxy.example.com:3128"))
+        opener.add_handler(ph)
+        password_manager = MockPasswordManager()
+        digest_handler = urllib.request.ProxyDigestAuthHandler(password_manager)
+        realm = "ACME Networks"
+        http_handler = MockHTTPHandlerRedirect(
+            407,
+            'Proxy-Authenticate: Digest realm="%s", nonce="abcdef0123456789", '
+            'qop="auth"\r\n\r\n' % realm)
+        opener.add_handler(digest_handler)
+        opener.add_handler(http_handler)
+
+        user, password = "wile", "coyote"
+        request_url = "http://acme.example.com/protected"
+        proxy_authority = "proxy.example.com:3128"
+
+        digest_handler.add_password(realm, proxy_authority, user, password)
+        self.assertEqual(realm, password_manager.realm)
+        self.assertEqual(proxy_authority, password_manager.url)
+
+        opener.open(request_url)
+
+        self.assertEqual(password_manager.target_realm, realm)
+        self.assertEqual(password_manager.target_url, proxy_authority)
+
+        self.assertEqual(len(http_handler.requests), 2)
+        self.assertFalse(
+            http_handler.requests[0].has_header("Proxy-authorization"))
+        self.assertTrue(
+            http_handler.requests[1].has_header("Proxy-authorization"))
+        self.assertTrue(
+            http_handler.requests[1]
+            .get_header("Proxy-authorization").startswith("Digest "))
+
+    def test_digest_auth_uses_full_url(self):
+        # The non-proxy case must keep using the full request URL (not
+        # just the host) for the password lookup, so that sub-URI realm
+        # matching (HTTPPasswordMgr.is_suburi) keeps working
+        opener = OpenerDirector()
+        password_manager = MockPasswordManager()
+        digest_handler = urllib.request.HTTPDigestAuthHandler(password_manager)
+        realm = "ACME Networks"
+        request_url = "http://acme.example.com/protected"
+        http_handler = MockHTTPHandlerRedirect(
+            401,
+            'WWW-Authenticate: Digest realm="%s", nonce="abcdef0123456789", '
+            'qop="auth"\r\n\r\n' % realm)
+        opener.add_handler(digest_handler)
+        opener.add_handler(http_handler)
+
+        digest_handler.add_password(realm, request_url, "wile", "coyote")
+        opener.open(request_url)
+
+        self.assertEqual(password_manager.target_realm, realm)
+        self.assertEqual(password_manager.target_url, request_url)
+
     def test_basic_and_digest_auth_handlers(self):
         # HTTPDigestAuthHandler raised an exception if it couldn't handle a 40*
         # response (https://bugs.python.org/issue1479302), where it should instead
@@ -1811,6 +1871,25 @@ def test_unsupported_auth_digest_handler(self):
         opener.add_handler(http_handler)
         self.assertRaises(ValueError, opener.open, "http://www.example.com")
 
+    def test_digest_auth_failure_code(self):
+        # gh-53852: a failed proxy digest auth must report 407, not 401.
+        headers = http.client.HTTPMessage()
+        for handler_class, auth_header, code in [
+                (urllib.request.HTTPDigestAuthHandler,
+                 'www-authenticate', 401),
+                (urllib.request.ProxyDigestAuthHandler,
+                 'proxy-authenticate', 407)]:
+            with self.subTest(handler=handler_class.__name__):
+                handler = handler_class()
+                handler.retried = 6
+                req = Request("http://www.example.com/")
+                with self.assertRaises(urllib.error.HTTPError) as cm:
+                    handler.http_error_auth_reqed(auth_header,
+                                                  "www.example.com",
+                                                  req, headers)
+                self.assertEqual(cm.exception.code, code)
+                cm.exception.close()
+
     def test_unsupported_auth_basic_handler(self):
         # While using BasicAuthHandler
         opener = OpenerDirector()

Lib/test/test_urllib2_localnet.py

@@ -351,6 +351,8 @@ def create_fake_proxy_handler(*args, **kwargs):
         self.server.start()
         self.server.ready.wait()
         proxy_url = "http://127.0.0.1:%d" % self.server.port
+        # Credentials must be registered against proxy authority
+        self.proxy_authority = "127.0.0.1:%d" % self.server.port
         handler = urllib.request.ProxyHandler({"http" : proxy_url})
         self.proxy_digest_handler = urllib.request.ProxyDigestAuthHandler()
         self.opener = urllib.request.build_opener(
@@ -361,29 +363,31 @@ def stop_server(self):
         self.server = None
 
     def test_proxy_with_bad_password_raises_httperror(self):
-        self.proxy_digest_handler.add_password(self.REALM, self.URL,
+        self.proxy_digest_handler.add_password(self.REALM, self.proxy_authority,
                                                self.USER, self.PASSWD+"bad")
         self.digest_auth_handler.set_qop("auth")
         with self.assertRaises(urllib.error.HTTPError) as cm:
             self.opener.open(self.URL)
+        self.assertEqual(cm.exception.code, 407)
         cm.exception.close()
 
     def test_proxy_with_no_password_raises_httperror(self):
         self.digest_auth_handler.set_qop("auth")
         with self.assertRaises(urllib.error.HTTPError) as cm:
             self.opener.open(self.URL)
+        self.assertEqual(cm.exception.code, 407)
         cm.exception.close()
 
     def test_proxy_qop_auth_works(self):
-        self.proxy_digest_handler.add_password(self.REALM, self.URL,
+        self.proxy_digest_handler.add_password(self.REALM, self.proxy_authority,
                                                self.USER, self.PASSWD)
         self.digest_auth_handler.set_qop("auth")
         with self.opener.open(self.URL) as result:
             while result.read():
                 pass
 
     def test_proxy_qop_auth_int_works_or_throws_urlerror(self):
-        self.proxy_digest_handler.add_password(self.REALM, self.URL,
+        self.proxy_digest_handler.add_password(self.REALM, self.proxy_authority,
                                                self.USER, self.PASSWD)
         self.digest_auth_handler.set_qop("auth-int")
         try:

Lib/urllib/request.py

@@ -1093,7 +1093,11 @@ def http_error_auth_reqed(self, auth_header, host, req, headers):
             # prompting for the information. Crap. This isn't great
             # but it's better than the current 'repeat until recursion
             # depth exceeded' approach <wink>
-            raise HTTPError(req.full_url, 401, "digest auth failed",
+            if auth_header.lower() == 'proxy-authenticate':
+                code = 407
+            else:
+                code = 401
+            raise HTTPError(req.full_url, code, "digest auth failed",
                             headers, None)
         else:
             self.retried += 1
@@ -1144,7 +1148,13 @@ def get_authorization(self, req, chal):
         if H is None:
             return None
 
-        user, pw = self.passwd.find_user_password(realm, req.full_url)
+        # proxy auth is looked up by the proxy's authority
+        # req.host, set by Request.set_proxy()
+        if self.auth_header == 'Proxy-Authorization':
+            authority = req.host
+        else:
+            authority = req.full_url
+        user, pw = self.passwd.find_user_password(realm, authority)
         if user is None:
             return None
 

Misc/NEWS.d/next/Library/2025-09-18-18-03-32.gh-issue-139105.1QQ_W4.rst

@@ -0,0 +1 @@
+Correct the EOF key shown in the :mod:`sqlite3` CLI ``.help`` text on Windows.

Misc/NEWS.d/next/Library/2026-09-21-21-07-12.gh-issue-157928.ZPlNYC.rst

@@ -0,0 +1 @@
+Correctly detect and handle OSC sequences in PyREPL.