Fix source decoding diagnostics and normalize decoded newlines (#9429)

Track coding-cookie locations, preserve codec initialization errors as SyntaxError, and normalize decoded source before parsing. Add unchanged CPython 3.15 source-encoding coverage via update_lib.

Assisted-by: OpenAI Codex: model version unavailable

Co-authored-by: Codex <codex@openai.com>

Lib/test/test_source_encoding.py

@@ -0,0 +1,559 @@
+# -*- coding: utf-8 -*-
+
+import unittest
+from test import support
+from test.support import script_helper
+from test.support.os_helper import TESTFN, unlink, rmtree
+from test.support.import_helper import unload
+import importlib
+import os
+import sys
+import subprocess
+import tempfile
+
+class MiscSourceEncodingTest(unittest.TestCase):
+
+    def test_import_encoded_module(self):
+        from test.encoded_modules import test_strings
+        # Make sure we're actually testing something
+        self.assertGreaterEqual(len(test_strings), 1)
+        for modname, encoding, teststr in test_strings:
+            mod = importlib.import_module('test.encoded_modules.'
+                                          'module_' + modname)
+            self.assertEqual(teststr, mod.test)
+
+    def test_compilestring(self):
+        # see #1882
+        c = compile(b"\n# coding: utf-8\nu = '\xc3\xb3'\n", "dummy", "exec")
+        d = {}
+        exec(c, d)
+        self.assertEqual(d['u'], '\xf3')
+
+    def test_issue2301(self):
+        try:
+            compile(b"# coding: cp932\nprint '\x94\x4e'", "dummy", "exec")
+        except SyntaxError as v:
+            self.assertEqual(v.text.rstrip('\n'), "print '\u5e74'")
+        else:
+            self.fail()
+
+    def test_issue4626(self):
+        c = compile("# coding=latin-1\n\u00c6 = '\u00c6'", "dummy", "exec")
+        d = {}
+        exec(c, d)
+        self.assertEqual(d['\xc6'], '\xc6')
+
+    def test_issue3297(self):
+        c = compile("a, b = '\U0001010F', '\\U0001010F'", "dummy", "exec")
+        d = {}
+        exec(c, d)
+        self.assertEqual(d['a'], d['b'])
+        self.assertEqual(len(d['a']), len(d['b']))
+        self.assertEqual(ascii(d['a']), ascii(d['b']))
+
+    def test_issue7820(self):
+        # Ensure that check_bom() restores all bytes in the right order if
+        # check_bom() fails in pydebug mode: a buffer starts with the first
+        # byte of a valid BOM, but next bytes are different
+
+        # one byte in common with the UTF-16-LE BOM
+        self.assertRaises(SyntaxError, eval, b'\xff\x20')
+
+        # one byte in common with the UTF-8 BOM
+        self.assertRaises(SyntaxError, eval, b'\xef\x20')
+
+        # two bytes in common with the UTF-8 BOM
+        self.assertRaises(SyntaxError, eval, b'\xef\xbb\x20')
+
+    def test_truncated_utf8_at_eof(self):
+        # Regression test for https://issues.oss-fuzz.com/issues/451112368
+        # Truncated multi-byte UTF-8 sequences at end of input caused an
+        # out-of-bounds read in Parser/tokenizer/helpers.c:valid_utf8().
+        truncated = [
+            b'\xc2',              # 2-byte lead, missing 1 continuation
+            b'\xdf',              # 2-byte lead, missing 1 continuation
+            b'\xe0',              # 3-byte lead, missing 2 continuations
+            b'\xe0\xa0',          # 3-byte lead, missing 1 continuation
+            b'\xf0\x90',          # 4-byte lead, missing 2 continuations
+            b'\xf0\x90\x80',      # 4-byte lead, missing 1 continuation
+            b'\xf3',              # 4-byte lead, missing 3 (the oss-fuzz reproducer)
+        ]
+        for seq in truncated:
+            with self.subTest(seq=seq):
+                self.assertRaises(SyntaxError, compile, seq, '<test>', 'exec')
+
+    @support.requires_subprocess()
+    def test_20731(self):
+        sub = subprocess.Popen([sys.executable,
+                        os.path.join(os.path.dirname(__file__),
+                                     'tokenizedata',
+                                     'coding20731.py')],
+                        stderr=subprocess.PIPE)
+        err = sub.communicate()[1]
+        self.assertEqual(sub.returncode, 0)
+        self.assertNotIn(b'SyntaxError', err)
+
+    def test_error_message(self):
+        compile(b'# -*- coding: iso-8859-15 -*-\n', 'dummy', 'exec')
+        compile(b'\xef\xbb\xbf\n', 'dummy', 'exec')
+        compile(b'\xef\xbb\xbf# -*- coding: utf-8 -*-\n', 'dummy', 'exec')
+        with self.assertRaisesRegex(SyntaxError, 'fake'):
+            compile(b'# -*- coding: fake -*-\n', 'dummy', 'exec')
+        with self.assertRaisesRegex(SyntaxError, 'iso-8859-15'):
+            compile(b'\xef\xbb\xbf# -*- coding: iso-8859-15 -*-\n',
+                    'dummy', 'exec')
+        with self.assertRaisesRegex(SyntaxError, 'BOM'):
+            compile(b'\xef\xbb\xbf# -*- coding: iso-8859-15 -*-\n',
+                    'dummy', 'exec')
+        with self.assertRaisesRegex(SyntaxError, 'fake'):
+            compile(b'\xef\xbb\xbf# -*- coding: fake -*-\n', 'dummy', 'exec')
+        with self.assertRaisesRegex(SyntaxError, 'BOM'):
+            compile(b'\xef\xbb\xbf# -*- coding: fake -*-\n', 'dummy', 'exec')
+
+    def test_bad_coding(self):
+        module_name = 'bad_coding'
+        self.verify_bad_module(module_name)
+
+    def test_bad_coding2(self):
+        module_name = 'bad_coding2'
+        self.verify_bad_module(module_name)
+
+    def verify_bad_module(self, module_name):
+        self.assertRaises(SyntaxError, __import__, 'test.tokenizedata.' + module_name)
+
+        path = os.path.dirname(__file__)
+        filename = os.path.join(path, 'tokenizedata', module_name + '.py')
+        with open(filename, "rb") as fp:
+            bytes = fp.read()
+        self.assertRaises(SyntaxError, compile, bytes, filename, 'exec')
+
+    def test_exec_valid_coding(self):
+        d = {}
+        exec(b'# coding: cp949\na = "\xaa\xa7"\n', d)
+        self.assertEqual(d['a'], '\u3047')
+
+    def test_file_parse(self):
+        # issue1134: all encodings outside latin-1 and utf-8 fail on
+        # multiline strings and long lines (>512 columns)
+        unload(TESTFN)
+        filename = TESTFN + ".py"
+        f = open(filename, "w", encoding="cp1252")
+        sys.path.insert(0, os.curdir)
+        try:
+            with f:
+                f.write("# -*- coding: cp1252 -*-\n")
+                f.write("'''A short string\n")
+                f.write("'''\n")
+                f.write("'A very long string %s'\n" % ("X" * 1000))
+
+            importlib.invalidate_caches()
+            __import__(TESTFN)
+        finally:
+            del sys.path[0]
+            unlink(filename)
+            unlink(filename + "c")
+            unlink(filename + "o")
+            unload(TESTFN)
+            rmtree('__pycache__')
+
+    def test_error_from_string(self):
+        # See http://bugs.python.org/issue6289
+        input = "# coding: ascii\n\N{SNOWMAN}".encode('utf-8')
+        with self.assertRaises(SyntaxError) as c:
+            compile(input, "<string>", "exec")
+        expected = "'ascii' codec can't decode byte 0xe2 in position 16: " \
+                   "ordinal not in range(128)"
+        self.assertStartsWith(c.exception.args[0], expected)
+
+    def test_file_parse_error_multiline(self):
+        # gh96611:
+        with open(TESTFN, "wb") as fd:
+            fd.write(b'print("""\n\xb1""")\n')
+
+        try:
+            retcode, stdout, stderr = script_helper.assert_python_failure(TESTFN)
+
+            self.assertGreater(retcode, 0)
+            self.assertIn(b"Non-UTF-8 code starting with '\\xb1'", stderr)
+        finally:
+            os.unlink(TESTFN)
+
+    def test_tokenizer_fstring_warning_in_first_line(self):
+        source = "0b1and 2"
+        with open(TESTFN, "w") as fd:
+            fd.write("{}".format(source))
+        try:
+            retcode, stdout, stderr = script_helper.assert_python_ok(TESTFN)
+            self.assertIn(b"SyntaxWarning: invalid binary litera", stderr)
+            self.assertEqual(stderr.count(source.encode()), 1)
+        finally:
+            os.unlink(TESTFN)
+
+
+BUFSIZ = 2**13
+
+class AbstractSourceEncodingTest:
+
+    def test_default_coding(self):
+        src = (b'print(ascii("\xc3\xa4"))\n')
+        self.check_script_output(src, br"'\xe4'")
+
+    def test_first_coding_line(self):
+        src = (b'#coding:iso8859-15\n'
+               b'print(ascii("\xc3\xa4"))\n')
+        self.check_script_output(src, br"'\xc3\u20ac'")
+
+    def test_second_coding_line(self):
+        src = (b'#!/usr/bin/python\n'
+               b'#coding:iso8859-15\n'
+               b'print(ascii("\xc3\xa4"))\n')
+        self.check_script_output(src, br"'\xc3\u20ac'")
+
+    def test_second_coding_line_empty_first_line(self):
+        src = (b'\n'
+               b'#coding:iso8859-15\n'
+               b'print(ascii("\xc3\xa4"))\n')
+        self.check_script_output(src, br"'\xc3\u20ac'")
+
+    def test_third_coding_line(self):
+        # Only first two lines are tested for a magic comment.
+        src = (b'#!/usr/bin/python\n'
+               b'#\n'
+               b'#coding:iso8859-15\n'
+               b'print(ascii("\xc3\xa4"))\n')
+        self.check_script_output(src, br"'\xe4'")
+
+    def test_double_coding_line(self):
+        # If the first line matches the second line is ignored.
+        src = (b'#coding:iso8859-15\n'
+               b'#coding:latin1\n'
+               b'print(ascii("\xc3\xa4"))\n')
+        self.check_script_output(src, br"'\xc3\u20ac'")
+
+    def test_double_coding_same_line(self):
+        src = (b'#coding:iso8859-15 coding:latin1\n'
+               b'print(ascii("\xc3\xa4"))\n')
+        self.check_script_output(src, br"'\xc3\u20ac'")
+
+    def test_double_coding_utf8(self):
+        src = (b'#coding:utf-8\n'
+               b'#coding:latin1\n'
+               b'print(ascii("\xc3\xa4"))\n')
+        self.check_script_output(src, br"'\xe4'")
+
+    def test_long_first_coding_line(self):
+        src = (b'#' + b' '*BUFSIZ + b'coding:iso8859-15\n'
+               b'print(ascii("\xc3\xa4"))\n')
+        self.check_script_output(src, br"'\xc3\u20ac'")
+
+    def test_long_second_coding_line(self):
+        src = (b'#!/usr/bin/python\n'
+               b'#' + b' '*BUFSIZ + b'coding:iso8859-15\n'
+               b'print(ascii("\xc3\xa4"))\n')
+        self.check_script_output(src, br"'\xc3\u20ac'")
+
+    def test_long_coding_line(self):
+        src = (b'#coding:iso-8859-15' + b' '*BUFSIZ + b'\n'
+               b'print(ascii("\xc3\xa4"))\n')
+        self.check_script_output(src, br"'\xc3\u20ac'")
+
+    def test_long_coding_name(self):
+        src = (b'#coding:iso-8859-1-' + b'x'*BUFSIZ + b'\n'
+               b'print(ascii("\xc3\xa4"))\n')
+        self.check_script_output(src, br"'\xc3\xa4'")
+
+    def test_long_first_utf8_line(self):
+        src = b'#' + b'\xc3\xa4'*(BUFSIZ//2) + b'\n'
+        self.check_script_output(src, b'')
+        src = b'# ' + b'\xc3\xa4'*(BUFSIZ//2) + b'\n'
+        self.check_script_output(src, b'')
+
+    def test_long_second_utf8_line(self):
+        src = b'\n#' + b'\xc3\xa4'*(BUFSIZ//2) + b'\n'
+        self.check_script_output(src, b'')
+        src = b'\n# ' + b'\xc3\xa4'*(BUFSIZ//2) + b'\n'
+        self.check_script_output(src, b'')
+
+    def test_first_non_utf8_coding_line(self):
+        src = (b'#coding:iso-8859-15 \xa4\n'
+               b'print(ascii("\xc3\xa4"))\n')
+        self.check_script_output(src, br"'\xc3\u20ac'")
+
+    def test_second_non_utf8_coding_line(self):
+        src = (b'#!/usr/bin/python\n'
+               b'#coding:iso-8859-15 \xa4\n'
+               b'print(ascii("\xc3\xa4"))\n')
+        self.check_script_output(src, br"'\xc3\u20ac'")
+
+    def test_first_utf8_coding_line_error(self):
+        src = (b'#coding:ascii \xc3\xa4\n'
+               b'raise RuntimeError\n')
+        self.check_script_error(src, br"(\(unicode error\) )?'ascii' codec can't decode byte")
+
+    def test_second_utf8_coding_line_error(self):
+        src = (b'#!/usr/bin/python\n'
+               b'#coding:ascii \xc3\xa4\n'
+               b'raise RuntimeError\n')
+        self.check_script_error(src, br"(\(unicode error\) )?'ascii' codec can't decode byte")
+
+    def test_utf8_bom(self):
+        src = (b'\xef\xbb\xbfprint(ascii("\xc3\xa4"))\n')
+        self.check_script_output(src, br"'\xe4'")
+
+    def test_utf8_bom_utf8_comments(self):
+        src = (b'\xef\xbb\xbf#\xc3\xa4\n'
+               b'#\xc3\xa4\n'
+               b'print(ascii("\xc3\xa4"))\n')
+        self.check_script_output(src, br"'\xe4'")
+
+    def test_utf8_bom_and_utf8_coding_line(self):
+        src = (b'\xef\xbb\xbf#coding:utf-8\n'
+               b'print(ascii("\xc3\xa4"))\n')
+        self.check_script_output(src, br"'\xe4'")
+
+    def test_utf8_bom_and_non_utf8_first_coding_line(self):
+        src = (b'\xef\xbb\xbf#coding:iso-8859-15\n'
+               b'raise RuntimeError\n')
+        self.check_script_error(src,
+                br"encoding problem: iso-8859-15 with BOM",
+                lineno=1)
+
+    def test_utf8_bom_and_non_utf8_second_coding_line(self):
+        src = (b'\xef\xbb\xbf#first\n'
+               b'#coding:iso-8859-15\n'
+               b'raise RuntimeError\n')
+        self.check_script_error(src,
+                br"encoding problem: iso-8859-15 with BOM",
+                lineno=2)
+
+    def test_non_utf8_shebang(self):
+        src = (b'#!/home/\xa4/bin/python\n'
+               b'#coding:iso-8859-15\n'
+               b'print(ascii("\xc3\xa4"))\n')
+        self.check_script_output(src, br"'\xc3\u20ac'")
+
+    def test_utf8_shebang_error(self):
+        src = (b'#!/home/\xc3\xa4/bin/python\n'
+               b'#coding:ascii\n'
+               b'raise RuntimeError\n')
+        self.check_script_error(src, br"(\(unicode error\) )?'ascii' codec can't decode byte")
+
+    def test_non_utf8_shebang_error(self):
+        src = (b'#!/home/\xa4/bin/python\n'
+               b'raise RuntimeError\n')
+        self.check_script_error(src, br"Non-UTF-8 code starting with .* on line 1",
+                                lineno=1)
+
+    def test_non_utf8_second_line_error(self):
+        src = (b'#first\n'
+               b'#second\xa4\n'
+               b'raise RuntimeError\n')
+        self.check_script_error(src,
+                br"Non-UTF-8 code starting with .* on line 2",
+                lineno=2)
+
+    def test_non_utf8_third_line_error(self):
+        src = (b'#first\n'
+               b'#second\n'
+               b'#third\xa4\n'
+               b'raise RuntimeError\n')
+        self.check_script_error(src,
+                br"Non-UTF-8 code starting with .* on line 3",
+                lineno=3)
+
+    def test_utf8_bom_non_utf8_third_line_error(self):
+        src = (b'\xef\xbb\xbf#first\n'
+               b'#second\n'
+               b'#third\xa4\n'
+               b'raise RuntimeError\n')
+        self.check_script_error(src,
+                br"Non-UTF-8 code starting with .* on line 3|"
+                br"'utf-8' codec can't decode byte",
+                lineno=3)
+
+    def test_utf_8_non_utf8_third_line_error(self):
+        src = (b'#coding: utf-8\n'
+               b'#second\n'
+               b'#third\xa4\n'
+               b'raise RuntimeError\n')
+        self.check_script_error(src,
+                br"Non-UTF-8 code starting with .* on line 3|"
+                br"'utf-8' codec can't decode byte",
+                lineno=3)
+
+    def test_utf8_non_utf8_third_line_error(self):
+        src = (b'#coding: utf8\n'
+               b'#second\n'
+               b'#third\xa4\n'
+               b'raise RuntimeError\n')
+        self.check_script_error(src,
+                br"'utf-8' codec can't decode byte")
+
+    def test_crlf(self):
+        src = (b'print(ascii("""\r\n"""))\n')
+        self.check_script_output(src, br"'\n'")
+
+    def test_crcrlf(self):
+        src = (b'print(ascii("""\r\r\n"""))\n')
+        self.check_script_output(src, br"'\n\n'")
+
+    def test_crcrcrlf(self):
+        src = (b'print(ascii("""\r\r\r\n"""))\n')
+        self.check_script_output(src, br"'\n\n\n'")
+
+    def test_crcrcrlf2(self):
+        src = (b'#coding:iso-8859-1\n'
+               b'print(ascii("""\r\r\r\n"""))\n')
+        self.check_script_output(src, br"'\n\n\n'")
+
+    def test_nul_in_first_coding_line(self):
+        src = (b'#coding:iso8859-15\x00\n'
+               b'\n'
+               b'\n'
+               b'raise RuntimeError\n')
+        self.check_script_error(src, br"source code (string )?cannot contain null bytes")
+
+    def test_nul_in_second_coding_line(self):
+        src = (b'#!/usr/bin/python\n'
+               b'#coding:iso8859-15\x00\n'
+               b'\n'
+               b'raise RuntimeError\n')
+        self.check_script_error(src, br"source code (string )?cannot contain null bytes")
+
+
+class UTF8ValidatorTest(unittest.TestCase):
+    @unittest.skipIf(not sys.platform.startswith("linux"),
+                     "Too slow to run on non-Linux platforms")
+    @support.requires_resource('cpu')
+    def test_invalid_utf8(self):
+        # This is a port of test_utf8_decode_invalid_sequences in
+        # test_unicode.py to exercise the separate utf8 validator in
+        # Parser/tokenizer/helpers.c used when reading source files.
+
+        # That file is written using low-level C file I/O, so the only way to
+        # test it is to write actual files to disk.
+
+        # Each example is put inside a string at the top of the file so
+        # it's an otherwise valid Python source file. Put some newlines
+        # beforehand so we can assert that the error is reported on the
+        # correct line.
+        template = b'\n\n\n"%s"\n'
+
+        fn = TESTFN
+        self.addCleanup(unlink, fn)
+
+        def check(content):
+            with open(fn, 'wb') as fp:
+                fp.write(template % content)
+            rc, stdout, stderr = script_helper.assert_python_failure(fn)
+            # We want to assert that the python subprocess failed gracefully,
+            # not via a signal.
+            self.assertGreaterEqual(rc, 1)
+            self.assertIn(b"Non-UTF-8 code starting with", stderr)
+            self.assertIn(b"on line 4", stderr)
+
+        # continuation bytes in a sequence of 2, 3, or 4 bytes
+        continuation_bytes = [bytes([x]) for x in range(0x80, 0xC0)]
+        # start bytes of a 2-byte sequence equivalent to code points < 0x7F
+        invalid_2B_seq_start_bytes = [bytes([x]) for x in range(0xC0, 0xC2)]
+        # start bytes of a 4-byte sequence equivalent to code points > 0x10FFFF
+        invalid_4B_seq_start_bytes = [bytes([x]) for x in range(0xF5, 0xF8)]
+        invalid_start_bytes = (
+            continuation_bytes + invalid_2B_seq_start_bytes +
+            invalid_4B_seq_start_bytes + [bytes([x]) for x in range(0xF7, 0x100)]
+        )
+
+        for byte in invalid_start_bytes:
+            check(byte)
+
+        for sb in invalid_2B_seq_start_bytes:
+            for cb in continuation_bytes:
+                check(sb + cb)
+
+        for sb in invalid_4B_seq_start_bytes:
+            for cb1 in continuation_bytes[:3]:
+                for cb3 in continuation_bytes[:3]:
+                    check(sb+cb1+b'\x80'+cb3)
+
+        for cb in [bytes([x]) for x in range(0x80, 0xA0)]:
+            check(b'\xE0'+cb+b'\x80')
+            check(b'\xE0'+cb+b'\xBF')
+            # surrogates
+        for cb in [bytes([x]) for x in range(0xA0, 0xC0)]:
+            check(b'\xED'+cb+b'\x80')
+            check(b'\xED'+cb+b'\xBF')
+        for cb in [bytes([x]) for x in range(0x80, 0x90)]:
+            check(b'\xF0'+cb+b'\x80\x80')
+            check(b'\xF0'+cb+b'\xBF\xBF')
+        for cb in [bytes([x]) for x in range(0x90, 0xC0)]:
+            check(b'\xF4'+cb+b'\x80\x80')
+            check(b'\xF4'+cb+b'\xBF\xBF')
+
+
+@support.force_not_colorized_test_class
+class BytesSourceEncodingTest(AbstractSourceEncodingTest, unittest.TestCase):
+
+    def check_script_output(self, src, expected):
+        with support.captured_stdout() as stdout:
+            exec(src)
+        out = stdout.getvalue().encode('latin1')
+        self.assertEqual(out.rstrip(), expected)
+
+    def check_script_error(self, src, expected, lineno=...):
+        with self.assertRaises(SyntaxError) as cm:
+            exec(src)
+        exc = cm.exception
+        self.assertRegex(str(exc), expected.decode())
+        if lineno is not ...:
+            self.assertEqual(exc.lineno, lineno)
+            line = src.splitlines()[lineno-1].decode(errors='replace')
+            if lineno == 1:
+                line = line.removeprefix('\ufeff')
+            self.assertEqual(line, exc.text)
+
+
+@support.force_not_colorized_test_class
+class FileSourceEncodingTest(AbstractSourceEncodingTest, unittest.TestCase):
+
+    def check_script_output(self, src, expected):
+        with tempfile.TemporaryDirectory() as tmpd:
+            fn = os.path.join(tmpd, 'test.py')
+            with open(fn, 'wb') as fp:
+                fp.write(src)
+            res = script_helper.assert_python_ok(fn)
+        self.assertEqual(res.out.rstrip(), expected)
+
+    def check_script_error(self, src, expected, lineno=...):
+        with tempfile.TemporaryDirectory() as tmpd:
+            fn = os.path.join(tmpd, 'test.py')
+            with open(fn, 'wb') as fp:
+                fp.write(src)
+            res = script_helper.assert_python_failure(fn)
+        err = res.err.rstrip()
+        self.assertRegex(err.splitlines()[-1], b'SyntaxError: ' + expected)
+        if lineno is not ...:
+            self.assertIn(f', line {lineno}\n'.encode(),
+                          err.replace(os.linesep.encode(), b'\n'))
+            line = src.splitlines()[lineno-1].decode(errors='replace')
+            if lineno == 1:
+                line = line.removeprefix('\ufeff')
+            self.assertIn(line.encode(), err)
+
+    def test_coding_spec_unknown_encoding(self):
+        src = (b'# coding: c1252\n'
+               b'print("Hi!")\n')
+        self.check_script_error(src, br"unknown encoding: c1252")
+
+    def test_coding_spec_decode_error(self):
+        src = (b'# coding: shift-jis\n'
+               b'print("\xc4\x85")\n')
+        self.check_script_error(src, br"'shift_jis' codec can't decode byte")
+
+    def test_coding_spec_non_text_encoding(self):
+        src = (b'# coding: hex_codec\n'
+               b'print("eggs")\n')
+        self.check_script_error(src, br"'hex_codec' is not a text encoding")
+
+
+if __name__ == "__main__":
+    unittest.main()

crates/vm/src/vm/compile.rs

@@ -144,9 +144,16 @@ impl CompileWarningError {
     }
 }
 
+#[cfg(feature = "parser")]
+struct SourceEncoding {
+    name: String,
+    lineno: usize,
+    line: core::ops::Range<usize>,
+}
+
 impl VirtualMachine {
     #[cfg(feature = "parser")]
-    fn detect_source_encoding(source: &[u8]) -> Option<String> {
+    fn detect_source_encoding(source: &[u8]) -> Option<SourceEncoding> {
         fn find_encoding_in_line(line: &[u8]) -> Option<String> {
             let hash_pos = line.iter().position(|&b| b == b'#')?;
             if !line[..hash_pos]
@@ -156,40 +163,60 @@ impl VirtualMachine {
                 return None;
             }
             let after_hash = &line[hash_pos..];
-            let coding_pos = after_hash.windows(6).position(|w| w == b"coding")?;
-            let after_coding = &after_hash[coding_pos + 6..];
-            let rest = if after_coding.first() == Some(&b':') || after_coding.first() == Some(&b'=')
-            {
-                &after_coding[1..]
-            } else {
-                return None;
-            };
-            let name: String = rest
-                .iter()
-                .copied()
-                .skip_while(|&b| b == b' ' || b == b'\t')
-                .take_while(|&b| b.is_ascii_alphanumeric() || b == b'-' || b == b'_' || b == b'.')
-                .map(|b| b as char)
-                .collect();
-            (!name.is_empty()).then(|| VirtualMachine::normalize_source_encoding(&name))
+            for (coding_pos, candidate) in after_hash.windows(6).enumerate() {
+                if candidate != b"coding" {
+                    continue;
+                }
+                let after_coding = &after_hash[coding_pos + 6..];
+                if !matches!(after_coding.first(), Some(b':' | b'=')) {
+                    continue;
+                }
+                let name: String = after_coding[1..]
+                    .iter()
+                    .copied()
+                    .skip_while(|&b| b == b' ' || b == b'\t')
+                    .take_while(|&b| {
+                        b.is_ascii_alphanumeric() || b == b'-' || b == b'_' || b == b'.'
+                    })
+                    .map(|b| b as char)
+                    .collect();
+                if !name.is_empty() {
+                    return Some(VirtualMachine::normalize_source_encoding(&name));
+                }
+            }
+            None
         }
 
-        let mut lines = source.splitn(3, |&b| b == b'\n');
-        if let Some(first) = lines.next() {
-            let first = first.strip_prefix(b"\xef\xbb\xbf").unwrap_or(first);
-            if let Some(enc) = find_encoding_in_line(first) {
-                return Some(enc);
-            }
-            let trimmed = first
+        let mut start = if source.starts_with(b"\xef\xbb\xbf") {
+            3
+        } else {
+            0
+        };
+        for lineno in 1..=2 {
+            let end = source[start..]
                 .iter()
-                .skip_while(|&&b| b == b' ' || b == b'\t' || b == b'\x0c' || b == b'\r')
-                .copied()
-                .collect::<Vec<_>>();
-            if !trimmed.is_empty() && trimmed[0] != b'#' {
+                .position(|&b| b == b'\r' || b == b'\n')
+                .map_or(source.len(), |offset| start + offset);
+            let line = &source[start..end];
+            if let Some(name) = find_encoding_in_line(line) {
+                return Some(SourceEncoding {
+                    name,
+                    lineno,
+                    line: start..end,
+                });
+            }
+            let first_nonspace = line
+                .iter()
+                .find(|&&b| b != b' ' && b != b'\t' && b != b'\x0c');
+            if first_nonspace.is_some_and(|&b| b != b'#') || end == source.len() {
                 return None;
             }
+            start = end + 1;
+            if source[end] == b'\r' && source.get(start) == Some(&b'\n') {
+                start += 1;
+            }
         }
-        lines.next().and_then(find_encoding_in_line)
+        None
     }
 
     #[cfg(feature = "parser")]
@@ -237,7 +264,7 @@ impl VirtualMachine {
             let decoded = {
                 #[cfg(feature = "parser")]
                 {
-                    let encoding = Self::detect_source_encoding(&buf);
+                    let encoding = Self::detect_source_encoding(&buf).map(|cookie| cookie.name);
                     if encoding.as_deref().is_none_or(Self::is_utf8_encoding) {
                         String::from_utf8_lossy(&buf).into_owned()
                     } else if encoding.as_deref() == Some("iso-8859-1") {
@@ -337,23 +364,28 @@ impl VirtualMachine {
         } else {
             Self::detect_source_encoding(source)
         };
-        let is_utf8 = encoding.as_deref().is_none_or(Self::is_utf8_encoding);
+        let encoding_name = encoding.as_ref().map(|cookie| cookie.name.as_str());
+        let is_utf8 = encoding_name.is_none_or(Self::is_utf8_encoding);
         if has_bom && !is_utf8 {
-            let enc = encoding.as_deref().unwrap_or("utf-8");
-            let after_bom = &source[3..];
-            let line_end = after_bom
+            let cookie = encoding.as_ref().unwrap();
+            let enc = cookie.name.as_str();
+            let line = &source[cookie.line.clone()];
+            let text = String::from_utf8_lossy(line).into_owned();
+            // CPython's string tokenizer includes the line terminator for
+            // the second coding line. Its range-error helper converts this
+            // byte column by counting non-continuation bytes in the prefix.
+            let end_byte_offset = line.len() + usize::from(cookie.lineno == 2);
+            let end_offset = 1 + line[..end_byte_offset - 1]
                 .iter()
-                .position(|&b| b == b'\n')
-                .unwrap_or(after_bom.len());
-            let text = String::from_utf8_lossy(&after_bom[..line_end]).into_owned();
-            let end_offset = text.chars().count() as i32;
+                .filter(|&&byte| byte & 0xc0 != 0x80)
+                .count();
             let msg = format!("encoding problem: {enc} with BOM");
             let location = self.ctx.new_tuple(vec![
                 self.ctx.new_str(filename).into(),
-                self.ctx.new_int(1).into(),
+                self.ctx.new_int(cookie.lineno).into(),
                 self.ctx.new_int(0).into(),
                 self.ctx.new_str(text).into(),
-                self.ctx.new_int(1).into(),
+                self.ctx.new_int(cookie.lineno).into(),
                 self.ctx.new_int(end_offset).into(),
             ]);
             return Err(self
@@ -366,31 +398,61 @@ impl VirtualMachine {
                 }));
         }
 
-        if is_utf8 {
+        let decoded = if is_utf8 {
             let src = if has_bom { &source[3..] } else { source };
             match core::str::from_utf8(src) {
-                Ok(s) => Ok(s.to_owned()),
-                Err(e) => Err(self.new_non_utf8_syntax_error(filename, src, e.valid_up_to())),
+                Ok(s) => s.to_owned(),
+                Err(e) => {
+                    return Err(self.new_non_utf8_syntax_error(filename, src, e.valid_up_to()));
+                }
             }
         } else {
-            let encoding = encoding.as_deref().unwrap();
+            let encoding = encoding_name.unwrap();
             let bytes = self.ctx.new_bytes(source.to_vec());
             let decoded = self
                 .state
                 .codec_registry
                 .decode_text(bytes.into(), encoding, None, self)
                 .map_err(|exc| {
-                    if exc.fast_isinstance(self.ctx.exceptions.lookup_error) {
-                        self.new_exception_msg(
-                            self.ctx.exceptions.syntax_error.to_owned(),
-                            format!("unknown encoding for '{filename}': {encoding}").into(),
+                    if exc.fast_isinstance(self.ctx.exceptions.syntax_error) {
+                        match exc
+                            .as_object()
+                            .set_attr("filename", self.ctx.new_str(filename), self)
+                        {
+                            Ok(()) => exc,
+                            Err(error) => error,
+                        }
+                    } else if exc.fast_isinstance(self.ctx.exceptions.lookup_error)
+                        || exc.fast_isinstance(self.ctx.exceptions.value_error)
+                    {
+                        let message = match exc.as_object().str(self) {
+                            Ok(message) => message,
+                            Err(error) => return error,
+                        };
+                        let location = self.ctx.new_tuple(vec![
+                            self.ctx.new_str(filename).into(),
+                            self.ctx.new_int(0).into(),
+                            self.ctx.new_int(-1).into(),
+                            self.ctx.none(),
+                        ]);
+                        self.invoke_exception(
+                            self.ctx.exceptions.syntax_error,
+                            vec![message.into(), location.into()],
                         )
+                        .unwrap_or_else(|error| error)
                     } else {
                         exc
                     }
                 })?;
-            Ok(decoded.to_string_lossy().into_owned())
-        }
+            decoded.to_string_lossy().into_owned()
+        };
+        // Normalize the source before parsing so literal values, warning ranges,
+        // and diagnostic text all refer to the same universal-newline input.
+        Ok(if decoded.contains('\r') {
+            decoded.replace("\r\n", "\n").replace('\r', "\n")
+        } else {
+            decoded
+        })
     }
 
     #[cfg(feature = "parser")]