Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/dulwich/patch.py: 10%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# patch.py -- For dealing with packed-style patches.
2# Copyright (C) 2009-2013 Jelmer Vernooij <jelmer@jelmer.uk>
3#
4# SPDX-License-Identifier: Apache-2.0 OR GPL-2.0-or-later
5# Dulwich is dual-licensed under the Apache License, Version 2.0 and the GNU
6# General Public License as published by the Free Software Foundation; version 2.0
7# or (at your option) any later version. You can redistribute it and/or
8# modify it under the terms of either of these two licenses.
9#
10# Unless required by applicable law or agreed to in writing, software
11# distributed under the License is distributed on an "AS IS" BASIS,
12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13# See the License for the specific language governing permissions and
14# limitations under the License.
15#
16# You should have received a copy of the licenses; if not, see
17# <http://www.gnu.org/licenses/> for a copy of the GNU General Public License
18# and <http://www.apache.org/licenses/LICENSE-2.0> for a copy of the Apache
19# License, Version 2.0.
20#
22"""Classes for dealing with git am-style patches.
24These patches are basically unified diffs with some extra metadata tacked
25on.
26"""
28__all__ = [
29 "DEFAULT_DIFF_ALGORITHM",
30 "FIRST_FEW_BYTES",
31 "DiffAlgorithmNotAvailable",
32 "FilePatch",
33 "MailinfoResult",
34 "PatchApplicationFailure",
35 "PatchHunk",
36 "apply_patch_hunks",
37 "apply_patches",
38 "commit_patch_id",
39 "gen_diff_header",
40 "get_summary",
41 "git_am_patch_split",
42 "git_base85_decode",
43 "is_binary",
44 "mailinfo",
45 "parse_patch_message",
46 "parse_unified_diff",
47 "patch_filename",
48 "patch_id",
49 "shortid",
50 "unified_diff",
51 "unified_diff_with_algorithm",
52 "write_blob_diff",
53 "write_commit_diff",
54 "write_commit_patch",
55 "write_object_diff",
56 "write_tree_diff",
57]
59import base64
60import email.message
61import email.parser
62import email.utils
63import os
64import re
65import stat
66import time
67import zlib
68from collections.abc import Generator, Sequence
69from dataclasses import dataclass
70from difflib import SequenceMatcher
71from typing import (
72 IO,
73 TYPE_CHECKING,
74 BinaryIO,
75 TextIO,
76)
78if TYPE_CHECKING:
79 from .config import Config
80 from .object_store import BaseObjectStore
81 from .repo import Repo
83from .objects import S_ISGITLINK, Blob, Commit, ObjectID, RawObjectID
85FIRST_FEW_BYTES = 8000
87DEFAULT_DIFF_ALGORITHM = "myers"
90class PatchApplicationFailure(Exception):
91 """Raised when a patch does not apply cleanly."""
94class DiffAlgorithmNotAvailable(Exception):
95 """Raised when a requested diff algorithm is not available."""
97 def __init__(self, algorithm: str, install_hint: str = "") -> None:
98 """Initialize exception.
100 Args:
101 algorithm: Name of the unavailable algorithm
102 install_hint: Optional installation hint
103 """
104 self.algorithm = algorithm
105 self.install_hint = install_hint
106 if install_hint:
107 super().__init__(
108 f"Diff algorithm '{algorithm}' requested but not available. {install_hint}"
109 )
110 else:
111 super().__init__(
112 f"Diff algorithm '{algorithm}' requested but not available."
113 )
116def write_commit_patch(
117 f: IO[bytes],
118 commit: "Commit",
119 contents: str | bytes,
120 progress: tuple[int, int],
121 version: str | None = None,
122 encoding: str | None = None,
123) -> None:
124 """Write a individual file patch.
126 Args:
127 f: File-like object to write to
128 commit: Commit object
129 contents: Contents of the patch
130 progress: tuple with current patch number and total.
131 version: Version string to include in patch header
132 encoding: Encoding to use for the patch
134 Returns:
135 tuple with filename and contents
136 """
137 encoding = encoding or getattr(f, "encoding", "ascii")
138 if encoding is None:
139 encoding = "ascii"
140 if isinstance(contents, str):
141 contents = contents.encode(encoding)
142 (num, total) = progress
143 f.write(
144 b"From "
145 + commit.id
146 + b" "
147 + time.ctime(commit.commit_time).encode(encoding)
148 + b"\n"
149 )
150 f.write(b"From: " + commit.author + b"\n")
151 f.write(
152 b"Date: " + time.strftime("%a, %d %b %Y %H:%M:%S %Z").encode(encoding) + b"\n"
153 )
154 # The first paragraph becomes the Subject, folded onto a single line; the
155 # rest of the message is written as the mail body.
156 subject, _, body = commit.message.partition(b"\n\n")
157 subject = subject.replace(b"\n", b" ").strip()
158 f.write((f"Subject: [PATCH {num}/{total}] ").encode(encoding) + subject + b"\n")
159 f.write(b"\n")
160 if body:
161 f.write(body)
162 if not body.endswith(b"\n"):
163 f.write(b"\n")
164 f.write(b"---\n")
165 try:
166 import subprocess
168 p = subprocess.Popen(
169 ["diffstat"], stdout=subprocess.PIPE, stdin=subprocess.PIPE
170 )
171 except (ImportError, OSError):
172 pass # diffstat not available?
173 else:
174 (diffstat, _) = p.communicate(contents)
175 f.write(diffstat)
176 f.write(b"\n")
177 f.write(contents)
178 f.write(b"-- \n")
179 if version is None:
180 from dulwich import __version__ as dulwich_version
182 f.write(b"Dulwich %d.%d.%d\n" % dulwich_version)
183 else:
184 if encoding is None:
185 encoding = "ascii"
186 f.write(version.encode(encoding) + b"\n")
189def _sanitize_subject_for_filename(text: str, max_length: int = 52) -> str:
190 """Sanitize a string for safe use as part of a filename.
192 Matches git's ``format_sanitized_subject`` behavior:
194 - Only ``[A-Za-z0-9._]`` are kept; other characters become ``-``
195 (collapsed across runs).
196 - Consecutive ``.`` are collapsed to a single ``.``.
197 - The result is truncated to ``max_length`` characters.
198 - Trailing ``.`` and ``-`` are stripped.
200 Args:
201 text: Input string (typically a commit subject line).
202 max_length: Maximum length of the returned string.
204 Returns: Sanitized string safe to embed in a filename.
205 """
206 result: list[str] = []
207 # 2 = initial, 1 = saw a non-title char, 0 = saw a title char
208 space = 2
209 i = 0
210 text_len = len(text)
211 while i < text_len:
212 c = text[i]
213 if ("A" <= c <= "Z") or ("a" <= c <= "z") or ("0" <= c <= "9") or c in "._":
214 if space == 1:
215 result.append("-")
216 space = 0
217 result.append(c)
218 if c == ".":
219 while i + 1 < text_len and text[i + 1] == ".":
220 i += 1
221 else:
222 space |= 1
223 i += 1
224 if len(result) >= max_length:
225 break
227 return "".join(result)[:max_length].rstrip(".-")
230def get_summary(commit: "Commit") -> str:
231 """Determine the summary line for use in a filename.
233 Sanitizes the commit subject so it is safe to use as a filename
234 component, matching git's ``format_sanitized_subject`` behavior:
235 characters outside ``[A-Za-z0-9._]`` are replaced with ``-`` (with
236 runs collapsed) and consecutive ``.`` are collapsed. The result is
237 also length-limited to prevent overly long filenames.
239 Args:
240 commit: Commit
241 Returns: Sanitized summary string suitable for use as a filename
242 component.
243 """
244 decoded = commit.message.decode(errors="replace")
245 lines = decoded.splitlines()
246 if not lines:
247 return ""
248 return _sanitize_subject_for_filename(lines[0])
251# Unified Diff
252def _format_range_unified(start: int, stop: int) -> str:
253 """Convert range to the "ed" format."""
254 # Per the diff spec at http://www.unix.org/single_unix_specification/
255 beginning = start + 1 # lines start numbering with one
256 length = stop - start
257 if length == 1:
258 return f"{beginning}"
259 if not length:
260 beginning -= 1 # empty ranges begin at line just before the range
261 return f"{beginning},{length}"
264def unified_diff(
265 a: Sequence[bytes],
266 b: Sequence[bytes],
267 fromfile: bytes = b"",
268 tofile: bytes = b"",
269 fromfiledate: str = "",
270 tofiledate: str = "",
271 n: int = 3,
272 lineterm: str = "\n",
273 tree_encoding: str = "utf-8",
274 output_encoding: str = "utf-8",
275) -> Generator[bytes, None, None]:
276 """difflib.unified_diff that can detect "No newline at end of file" as original "git diff" does.
278 Based on the same function in Python2.7 difflib.py
279 """
280 started = False
281 for group in SequenceMatcher(a=a, b=b).get_grouped_opcodes(n):
282 if not started:
283 started = True
284 fromdate = f"\t{fromfiledate}" if fromfiledate else ""
285 todate = f"\t{tofiledate}" if tofiledate else ""
286 yield f"--- {fromfile.decode(tree_encoding)}{fromdate}{lineterm}".encode(
287 output_encoding
288 )
289 yield f"+++ {tofile.decode(tree_encoding)}{todate}{lineterm}".encode(
290 output_encoding
291 )
293 first, last = group[0], group[-1]
294 file1_range = _format_range_unified(first[1], last[2])
295 file2_range = _format_range_unified(first[3], last[4])
296 yield f"@@ -{file1_range} +{file2_range} @@{lineterm}".encode(output_encoding)
298 for tag, i1, i2, j1, j2 in group:
299 if tag == "equal":
300 for line in a[i1:i2]:
301 yield b" " + line
302 continue
303 if tag in ("replace", "delete"):
304 for line in a[i1:i2]:
305 if not line[-1:] == b"\n":
306 line += b"\n\\ No newline at end of file\n"
307 yield b"-" + line
308 if tag in ("replace", "insert"):
309 for line in b[j1:j2]:
310 if not line[-1:] == b"\n":
311 line += b"\n\\ No newline at end of file\n"
312 yield b"+" + line
315def _get_sequence_matcher(
316 algorithm: str, a: Sequence[bytes], b: Sequence[bytes]
317) -> SequenceMatcher[bytes]:
318 """Get appropriate sequence matcher for the given algorithm.
320 Args:
321 algorithm: Diff algorithm ("myers" or "patience")
322 a: First sequence
323 b: Second sequence
325 Returns:
326 Configured sequence matcher instance
328 Raises:
329 DiffAlgorithmNotAvailable: If patience requested but not available
330 """
331 if algorithm == "patience":
332 try:
333 from patiencediff import PatienceSequenceMatcher
335 return PatienceSequenceMatcher(None, a, b) # type: ignore[no-any-return,unused-ignore]
336 except ImportError:
337 raise DiffAlgorithmNotAvailable(
338 "patience", "Install with: pip install 'dulwich[patiencediff]'"
339 )
340 else:
341 return SequenceMatcher(a=a, b=b)
344def unified_diff_with_algorithm(
345 a: Sequence[bytes],
346 b: Sequence[bytes],
347 fromfile: bytes = b"",
348 tofile: bytes = b"",
349 fromfiledate: str = "",
350 tofiledate: str = "",
351 n: int = 3,
352 lineterm: str = "\n",
353 tree_encoding: str = "utf-8",
354 output_encoding: str = "utf-8",
355 algorithm: str | None = None,
356) -> Generator[bytes, None, None]:
357 """Generate unified diff with specified algorithm.
359 Args:
360 a: First sequence of lines
361 b: Second sequence of lines
362 fromfile: Name of first file
363 tofile: Name of second file
364 fromfiledate: Date of first file
365 tofiledate: Date of second file
366 n: Number of context lines
367 lineterm: Line terminator
368 tree_encoding: Encoding for tree paths
369 output_encoding: Encoding for output
370 algorithm: Diff algorithm to use ("myers" or "patience")
372 Returns:
373 Generator yielding diff lines
375 Raises:
376 DiffAlgorithmNotAvailable: If patience algorithm requested but patiencediff not available
377 """
378 if algorithm is None:
379 algorithm = DEFAULT_DIFF_ALGORITHM
381 matcher = _get_sequence_matcher(algorithm, a, b)
383 started = False
384 for group in matcher.get_grouped_opcodes(n):
385 if not started:
386 started = True
387 fromdate = f"\t{fromfiledate}" if fromfiledate else ""
388 todate = f"\t{tofiledate}" if tofiledate else ""
389 yield f"--- {fromfile.decode(tree_encoding)}{fromdate}{lineterm}".encode(
390 output_encoding
391 )
392 yield f"+++ {tofile.decode(tree_encoding)}{todate}{lineterm}".encode(
393 output_encoding
394 )
396 first, last = group[0], group[-1]
397 file1_range = _format_range_unified(first[1], last[2])
398 file2_range = _format_range_unified(first[3], last[4])
399 yield f"@@ -{file1_range} +{file2_range} @@{lineterm}".encode(output_encoding)
401 for tag, i1, i2, j1, j2 in group:
402 if tag == "equal":
403 for line in a[i1:i2]:
404 yield b" " + line
405 continue
406 if tag in ("replace", "delete"):
407 for line in a[i1:i2]:
408 if not line[-1:] == b"\n":
409 line += b"\n\\ No newline at end of file\n"
410 yield b"-" + line
411 if tag in ("replace", "insert"):
412 for line in b[j1:j2]:
413 if not line[-1:] == b"\n":
414 line += b"\n\\ No newline at end of file\n"
415 yield b"+" + line
418def is_binary(content: bytes) -> bool:
419 """See if the first few bytes contain any null characters.
421 Args:
422 content: Bytestring to check for binary content
423 """
424 return b"\0" in content[:FIRST_FEW_BYTES]
427def shortid(hexsha: bytes | None) -> bytes:
428 """Get short object ID.
430 Args:
431 hexsha: Full hex SHA or None
433 Returns:
434 7-character short ID
435 """
436 if hexsha is None:
437 return b"0" * 7
438 else:
439 return hexsha[:7]
442def patch_filename(p: bytes | None, root: bytes) -> bytes:
443 """Generate patch filename.
445 Args:
446 p: Path or None
447 root: Root directory
449 Returns:
450 Full patch filename
451 """
452 if p is None:
453 return b"/dev/null"
454 else:
455 return root + b"/" + p
458def write_object_diff(
459 f: IO[bytes],
460 store: "BaseObjectStore",
461 old_file: tuple[bytes | None, int | None, ObjectID | None],
462 new_file: tuple[bytes | None, int | None, ObjectID | None],
463 diff_binary: bool = False,
464 diff_algorithm: str | None = None,
465) -> None:
466 """Write the diff for an object.
468 Args:
469 f: File-like object to write to
470 store: Store to retrieve objects from, if necessary
471 old_file: (path, mode, hexsha) tuple
472 new_file: (path, mode, hexsha) tuple
473 diff_binary: Whether to diff files even if they
474 are considered binary files by is_binary().
475 diff_algorithm: Algorithm to use for diffing ("myers" or "patience")
477 Note: the tuple elements should be None for nonexistent files
478 """
479 (old_path, old_mode, old_id) = old_file
480 (new_path, new_mode, new_id) = new_file
481 patched_old_path = patch_filename(old_path, b"a")
482 patched_new_path = patch_filename(new_path, b"b")
484 def content(mode: int | None, hexsha: ObjectID | None) -> Blob:
485 """Get blob content for a file.
487 Args:
488 mode: File mode
489 hexsha: Object SHA
491 Returns:
492 Blob object
493 """
494 if hexsha is None:
495 return Blob.from_string(b"")
496 elif mode is not None and S_ISGITLINK(mode):
497 return Blob.from_string(b"Subproject commit " + hexsha + b"\n")
498 else:
499 obj = store[hexsha]
500 if isinstance(obj, Blob):
501 return obj
502 else:
503 # Fallback for non-blob objects
504 return Blob.from_string(obj.as_raw_string())
506 def lines(content: "Blob") -> list[bytes]:
507 """Split blob content into lines.
509 Args:
510 content: Blob content
512 Returns:
513 List of lines
514 """
515 if not content:
516 return []
517 else:
518 return content.splitlines()
520 f.writelines(
521 gen_diff_header((old_path, new_path), (old_mode, new_mode), (old_id, new_id))
522 )
523 old_content = content(old_mode, old_id)
524 new_content = content(new_mode, new_id)
525 if not diff_binary and (is_binary(old_content.data) or is_binary(new_content.data)):
526 binary_diff = (
527 b"Binary files "
528 + patched_old_path
529 + b" and "
530 + patched_new_path
531 + b" differ\n"
532 )
533 f.write(binary_diff)
534 else:
535 f.writelines(
536 unified_diff_with_algorithm(
537 lines(old_content),
538 lines(new_content),
539 patched_old_path,
540 patched_new_path,
541 algorithm=diff_algorithm,
542 )
543 )
546# TODO(jelmer): Support writing unicode, rather than bytes.
547def gen_diff_header(
548 paths: tuple[bytes | None, bytes | None],
549 modes: tuple[int | None, int | None],
550 shas: tuple[bytes | None, bytes | None],
551) -> Generator[bytes, None, None]:
552 """Write a blob diff header.
554 Args:
555 paths: Tuple with old and new path
556 modes: Tuple with old and new modes
557 shas: Tuple with old and new shas
558 """
559 (old_path, new_path) = paths
560 (old_mode, new_mode) = modes
561 (old_sha, new_sha) = shas
562 if old_path is None and new_path is not None:
563 old_path = new_path
564 if new_path is None and old_path is not None:
565 new_path = old_path
566 old_path = patch_filename(old_path, b"a")
567 new_path = patch_filename(new_path, b"b")
568 yield b"diff --git " + old_path + b" " + new_path + b"\n"
570 if old_mode != new_mode:
571 if new_mode is not None:
572 if old_mode is not None:
573 yield (f"old file mode {old_mode:o}\n").encode("ascii")
574 yield (f"new file mode {new_mode:o}\n").encode("ascii")
575 else:
576 yield (f"deleted file mode {old_mode:o}\n").encode("ascii")
577 yield b"index " + shortid(old_sha) + b".." + shortid(new_sha)
578 if new_mode is not None and old_mode is not None:
579 yield (f" {new_mode:o}").encode("ascii")
580 yield b"\n"
583# TODO(jelmer): Support writing unicode, rather than bytes.
584def write_blob_diff(
585 f: IO[bytes],
586 old_file: tuple[bytes | None, int | None, "Blob | None"],
587 new_file: tuple[bytes | None, int | None, "Blob | None"],
588 diff_algorithm: str | None = None,
589) -> None:
590 """Write blob diff.
592 Args:
593 f: File-like object to write to
594 old_file: (path, mode, hexsha) tuple (None if nonexisting)
595 new_file: (path, mode, hexsha) tuple (None if nonexisting)
596 diff_algorithm: Algorithm to use for diffing ("myers" or "patience")
598 Note: The use of write_object_diff is recommended over this function.
599 """
600 (old_path, old_mode, old_blob) = old_file
601 (new_path, new_mode, new_blob) = new_file
602 patched_old_path = patch_filename(old_path, b"a")
603 patched_new_path = patch_filename(new_path, b"b")
605 def lines(blob: "Blob | None") -> list[bytes]:
606 """Split blob content into lines.
608 Args:
609 blob: Blob object or None
611 Returns:
612 List of lines
613 """
614 if blob is not None:
615 return blob.splitlines()
616 else:
617 return []
619 f.writelines(
620 gen_diff_header(
621 (old_path, new_path),
622 (old_mode, new_mode),
623 (getattr(old_blob, "id", None), getattr(new_blob, "id", None)),
624 )
625 )
626 old_contents = lines(old_blob)
627 new_contents = lines(new_blob)
628 f.writelines(
629 unified_diff_with_algorithm(
630 old_contents,
631 new_contents,
632 patched_old_path,
633 patched_new_path,
634 algorithm=diff_algorithm,
635 )
636 )
639def write_tree_diff(
640 f: IO[bytes],
641 store: "BaseObjectStore",
642 old_tree: ObjectID | None,
643 new_tree: ObjectID | None,
644 diff_binary: bool = False,
645 diff_algorithm: str | None = None,
646) -> None:
647 """Write tree diff.
649 Args:
650 f: File-like object to write to.
651 store: Object store to read from
652 old_tree: Old tree id
653 new_tree: New tree id
654 diff_binary: Whether to diff files even if they
655 are considered binary files by is_binary().
656 diff_algorithm: Algorithm to use for diffing ("myers" or "patience")
657 """
658 changes = store.tree_changes(old_tree, new_tree)
659 for (oldpath, newpath), (oldmode, newmode), (oldsha, newsha) in changes:
660 write_object_diff(
661 f,
662 store,
663 (oldpath, oldmode, oldsha),
664 (newpath, newmode, newsha),
665 diff_binary=diff_binary,
666 diff_algorithm=diff_algorithm,
667 )
670def write_commit_diff(
671 f: IO[bytes],
672 store: "BaseObjectStore",
673 commit: "Commit",
674 diff_binary: bool = False,
675 diff_algorithm: str | None = None,
676) -> None:
677 """Write the diff a commit introduces against its first parent.
679 Args:
680 f: File-like object to write to.
681 store: Object store to read from.
682 commit: Commit whose changes to diff. Root commits (with no parents)
683 are diffed against the empty tree.
684 diff_binary: Whether to diff files even if they are considered binary
685 files by is_binary().
686 diff_algorithm: Algorithm to use for diffing ("myers" or "patience").
687 """
688 if commit.parents:
689 parent = store[commit.parents[0]]
690 assert isinstance(parent, Commit)
691 parent_tree: ObjectID | None = parent.tree
692 else:
693 parent_tree = None
694 write_tree_diff(
695 f,
696 store,
697 parent_tree,
698 commit.tree,
699 diff_binary=diff_binary,
700 diff_algorithm=diff_algorithm,
701 )
704def git_am_patch_split(
705 f: TextIO | BinaryIO, encoding: str | None = None
706) -> tuple["Commit", bytes, bytes | None]:
707 """Parse a git-am-style patch and split it up into bits.
709 Args:
710 f: File-like object to parse
711 encoding: Encoding to use when creating Git objects
712 Returns: Tuple with commit object, diff contents and git version
713 """
714 encoding = encoding or getattr(f, "encoding", "ascii")
715 encoding = encoding or "ascii"
716 contents = f.read()
717 if isinstance(contents, bytes):
718 bparser = email.parser.BytesParser()
719 msg = bparser.parsebytes(contents)
720 else:
721 uparser = email.parser.Parser()
722 msg = uparser.parsestr(contents)
723 return parse_patch_message(msg, encoding)
726def parse_patch_message(
727 msg: email.message.Message, encoding: str | None = None
728) -> tuple["Commit", bytes, bytes | None]:
729 """Extract a Commit object and patch from an e-mail message.
731 Args:
732 msg: An email message (email.message.Message)
733 encoding: Encoding to use to encode Git commits
734 Returns: Tuple with commit object, diff contents and git version
735 """
736 c = Commit()
737 if encoding is None:
738 encoding = "ascii"
739 c.author = msg["from"].encode(encoding)
740 c.committer = msg["from"].encode(encoding)
741 try:
742 patch_tag_start = msg["subject"].index("[PATCH")
743 except ValueError:
744 subject = msg["subject"]
745 else:
746 close = msg["subject"].index("] ", patch_tag_start)
747 subject = msg["subject"][close + 2 :]
748 c.message = (subject.replace("\n", "") + "\n").encode(encoding)
749 first = True
751 body = msg.get_payload(decode=True)
752 if isinstance(body, str):
753 body = body.encode(encoding)
754 if isinstance(body, bytes):
755 lines = body.splitlines(True)
756 else:
757 # Handle other types by converting to string first
758 lines = str(body).encode(encoding).splitlines(True)
759 line_iter = iter(lines)
761 for line in line_iter:
762 if line == b"---\n":
763 break
764 if first:
765 if line.startswith(b"From: "):
766 c.author = line[len(b"From: ") :].rstrip()
767 else:
768 c.message += b"\n" + line
769 first = False
770 else:
771 c.message += line
772 diff = b""
773 for line in line_iter:
774 if line == b"-- \n":
775 break
776 diff += line
777 try:
778 version = next(line_iter).rstrip(b"\n")
779 except StopIteration:
780 version = None
781 return c, diff, version
784def patch_id(diff_data: bytes) -> bytes:
785 """Compute patch ID for a diff.
787 The patch ID is computed by normalizing the diff and computing a SHA1 hash.
788 This follows git's patch-id algorithm which:
789 1. Removes whitespace from lines starting with + or -
790 2. Replaces line numbers in @@ headers with a canonical form
791 3. Computes SHA1 of the result
793 Args:
794 diff_data: Raw diff data as bytes
796 Returns:
797 SHA1 hash of normalized diff (40-byte hex string)
799 TODO: This implementation uses a simple line-by-line approach. For better
800 compatibility with git's patch-id, consider using proper patch parsing that:
801 - Handles edge cases in diff format (binary diffs, mode changes, etc.)
802 - Properly parses unified diff format according to the spec
803 - Matches git's exact normalization algorithm byte-for-byte
804 See git's patch-id.c for reference implementation.
805 """
806 import hashlib
807 import re
809 # Normalize the diff for patch-id computation
810 normalized_lines = []
812 for line in diff_data.split(b"\n"):
813 # Skip diff headers (diff --git, index, ---, +++)
814 if line.startswith(
815 (
816 b"diff --git ",
817 b"index ",
818 b"--- ",
819 b"+++ ",
820 b"new file mode ",
821 b"old file mode ",
822 b"deleted file mode ",
823 b"new mode ",
824 b"old mode ",
825 b"similarity index ",
826 b"dissimilarity index ",
827 b"rename from ",
828 b"rename to ",
829 b"copy from ",
830 b"copy to ",
831 )
832 ):
833 continue
835 # Normalize @@ headers to a canonical form
836 if line.startswith(b"@@"):
837 # Replace line numbers with canonical form
838 match = re.match(rb"^@@\s+-\d+(?:,\d+)?\s+\+\d+(?:,\d+)?\s+@@", line)
839 if match:
840 # Use canonical hunk header without line numbers
841 normalized_lines.append(b"@@")
842 continue
844 # For +/- lines, strip all whitespace
845 if line.startswith((b"+", b"-")):
846 # Keep the +/- prefix but remove all whitespace from the rest
847 if len(line) > 1:
848 # Remove all whitespace from the content
849 content = line[1:].replace(b" ", b"").replace(b"\t", b"")
850 normalized_lines.append(line[:1] + content)
851 else:
852 # Just +/- alone
853 normalized_lines.append(line[:1])
854 continue
856 # Keep context lines and other content as-is
857 if line.startswith(b" ") or line == b"":
858 normalized_lines.append(line)
860 # Join normalized lines and compute SHA1
861 normalized = b"\n".join(normalized_lines)
862 return hashlib.sha1(normalized).hexdigest().encode("ascii")
865def commit_patch_id(
866 store: "BaseObjectStore", commit_id: ObjectID | RawObjectID
867) -> bytes:
868 """Compute patch ID for a commit.
870 Args:
871 store: Object store to read objects from
872 commit_id: Commit ID (40-byte hex string)
874 Returns:
875 Patch ID (40-byte hex string)
876 """
877 from io import BytesIO
879 commit = store[commit_id]
880 assert isinstance(commit, Commit)
882 diff_output = BytesIO()
883 write_commit_diff(diff_output, store, commit)
885 return patch_id(diff_output.getvalue())
888@dataclass
889class MailinfoResult:
890 """Result of mailinfo parsing.
892 Attributes:
893 author_name: Author's name
894 author_email: Author's email address
895 author_date: Author's date (if present in the email)
896 subject: Processed subject line
897 message: Commit message body
898 patch: Patch content
899 message_id: Message-ID header (if -m/--message-id was used)
900 """
902 author_name: str
903 author_email: str
904 author_date: str | None
905 subject: str
906 message: str
907 patch: str
908 message_id: str | None = None
911def _munge_subject(subject: str, keep_subject: bool, keep_non_patch: bool) -> str:
912 """Munge email subject line for commit message.
914 Args:
915 subject: Original subject line
916 keep_subject: If True, keep subject intact (-k option)
917 keep_non_patch: If True, only strip [PATCH] (-b option)
919 Returns:
920 Processed subject line
921 """
922 if keep_subject:
923 return subject
925 result = subject
927 # First remove Re: prefixes (they can appear before brackets)
928 while True:
929 new_result = re.sub(r"^\s*(?:re|RE|Re):\s*", "", result, flags=re.IGNORECASE)
930 if new_result == result:
931 break
932 result = new_result
934 # Remove bracketed strings
935 if keep_non_patch:
936 # Only remove brackets containing "PATCH"
937 # Match each bracket individually anywhere in the string
938 while True:
939 # Remove PATCH bracket, but be careful with whitespace
940 new_result = re.sub(
941 r"\[[^\]]*?PATCH[^\]]*?\](\s+)?", r"\1", result, flags=re.IGNORECASE
942 )
943 if new_result == result:
944 break
945 result = new_result
946 else:
947 # Remove all bracketed strings
948 while True:
949 new_result = re.sub(r"^\s*\[.*?\]\s*", "", result)
950 if new_result == result:
951 break
952 result = new_result
954 # Remove leading/trailing whitespace
955 result = result.strip()
957 # Normalize multiple whitespace to single space
958 result = re.sub(r"\s+", " ", result)
960 return result
963def _find_scissors_line(lines: list[bytes]) -> int | None:
964 """Find the scissors line in message body.
966 Args:
967 lines: List of lines in the message body
969 Returns:
970 Index of scissors line, or None if not found
971 """
972 # The perforation alternatives are written so no two unbounded ``-+``
973 # quantifiers are separated by an optional/empty subpattern. The earlier
974 # form ``(?:>?\s*-+\s*)?(?:8<|>8)?\s*-+\s*`` let a single run of dashes be
975 # split between two ``-+`` groups, so a long non-matching line of dashes
976 # backtracked quadratically (ReDoS) when fed through ``mailinfo`` /
977 # ``git am --scissors`` on an untrusted mailbox. Here a marker is the
978 # mandatory pivot between the two dash runs, and the marker-less case uses a
979 # bounded ``(?:\s+-+)?`` second run, keeping the match linear.
980 scissors_pattern = re.compile(
981 rb"^>?\s*-+(?:\s+-+)?\s*$"
982 rb"|^>?\s*(?:-+\s*)?(?:8<|>8)\s*-+\s*$"
983 rb"|^(?:>?\s*-+\s*)(?:cut here|scissors)(?:\s*-+)?$",
984 re.IGNORECASE,
985 )
987 for i, line in enumerate(lines):
988 if scissors_pattern.match(line.strip()):
989 return i
991 return None
994def git_base85_decode(data: bytes) -> bytes:
995 """Decode Git's base85-encoded binary data.
997 Each line starts with a byte giving the decoded length of that line
998 (``A``-``Z`` for 1-26, ``a``-``z`` for 27-52), followed by groups of five
999 base85 characters that each encode four bytes. Git's alphabet is the
1000 RFC 1924 one used by ``base64.b85decode``.
1002 Args:
1003 data: Base85-encoded data as bytes (may contain multiple lines)
1005 Returns:
1006 Decoded binary data
1008 Raises:
1009 ValueError: If the data is invalid
1010 """
1011 result = bytearray()
1012 for line in data.splitlines():
1013 if not line:
1014 continue
1015 length_byte = line[0]
1016 if ord("A") <= length_byte <= ord("Z"):
1017 length = length_byte - ord("A") + 1
1018 elif ord("a") <= length_byte <= ord("z"):
1019 length = length_byte - ord("a") + 27
1020 else:
1021 raise ValueError(f"Invalid base85 line length byte: {line[:1]!r}")
1022 encoded = line[1:]
1023 if len(encoded) != (length + 3) // 4 * 5:
1024 raise ValueError(
1025 f"Base85 line has {len(encoded)} characters, "
1026 f"expected {(length + 3) // 4 * 5} for {length} bytes"
1027 )
1028 result.extend(base64.b85decode(encoded)[:length])
1029 return bytes(result)
1032@dataclass
1033class PatchHunk:
1034 """Represents a single hunk in a unified diff.
1036 Attributes:
1037 old_start: Starting line number in old file
1038 old_count: Number of lines in old file
1039 new_start: Starting line number in new file
1040 new_count: Number of lines in new file
1041 lines: List of diff lines (prefixed with ' ', '+', or '-')
1042 """
1044 old_start: int
1045 old_count: int
1046 new_start: int
1047 new_count: int
1048 lines: list[bytes]
1051@dataclass
1052class FilePatch:
1053 """Represents a patch for a single file.
1055 Attributes:
1056 old_path: Path to old file (None for new files)
1057 new_path: Path to new file (None for deleted files)
1058 old_mode: Mode of old file (None for new files)
1059 new_mode: Mode of new file (None for deleted files)
1060 hunks: List of PatchHunk objects
1061 binary: True if this is a binary patch
1062 rename_from: Original path for renames (None if not a rename)
1063 rename_to: New path for renames (None if not a rename)
1064 copy_from: Source path for copies (None if not a copy)
1065 copy_to: Destination path for copies (None if not a copy)
1066 binary_old: Reverse binary data for binary patches (base85 encoded)
1067 binary_new: Forward binary data for binary patches (base85 encoded)
1068 binary_old_delta: True if binary_old is a delta against the new
1069 content rather than a literal copy of the old content
1070 binary_new_delta: True if binary_new is a delta against the old
1071 content rather than a literal copy of the new content
1072 """
1074 old_path: bytes | None
1075 new_path: bytes | None
1076 old_mode: int | None
1077 new_mode: int | None
1078 hunks: list[PatchHunk]
1079 binary: bool = False
1080 rename_from: bytes | None = None
1081 rename_to: bytes | None = None
1082 copy_from: bytes | None = None
1083 copy_to: bytes | None = None
1084 binary_old: bytes | None = None
1085 binary_new: bytes | None = None
1086 binary_old_delta: bool = False
1087 binary_new_delta: bool = False
1090_C_STYLE_ESCAPES = {
1091 ord("a"): b"\a",
1092 ord("b"): b"\b",
1093 ord("t"): b"\t",
1094 ord("n"): b"\n",
1095 ord("v"): b"\v",
1096 ord("f"): b"\f",
1097 ord("r"): b"\r",
1098 ord('"'): b'"',
1099 ord("\\"): b"\\",
1100}
1103def _unquote_c_style(text: bytes) -> tuple[bytes, bytes]:
1104 """Unquote a C-style quoted name at the start of ``text``.
1106 Git quotes names containing special or non-ASCII characters this way.
1108 Returns:
1109 Tuple with the unquoted name and the remainder of ``text`` after the
1110 closing quote.
1112 Raises:
1113 ValueError: If the quoted name is malformed
1114 """
1115 if not text.startswith(b'"'):
1116 raise ValueError(f"Name is not quoted: {text!r}")
1117 result = bytearray()
1118 i = 1
1119 while i < len(text):
1120 c = text[i]
1121 if c == ord('"'):
1122 return bytes(result), text[i + 1 :]
1123 if c != ord("\\"):
1124 result.append(c)
1125 i += 1
1126 continue
1127 escaped = text[i + 1 : i + 2]
1128 if escaped and escaped[0] in _C_STYLE_ESCAPES:
1129 result += _C_STYLE_ESCAPES[escaped[0]]
1130 i += 2
1131 elif re.fullmatch(rb"[0-3][0-7][0-7]", text[i + 1 : i + 4]):
1132 result.append(int(text[i + 1 : i + 4], 8))
1133 i += 4
1134 else:
1135 raise ValueError(f"Invalid escape in quoted name: {text!r}")
1136 raise ValueError(f"Unterminated quoted name: {text!r}")
1139def _unquote_name(name: bytes) -> bytes:
1140 """Unquote a name from a patch header if git quoted it."""
1141 if not name.startswith(b'"'):
1142 return name
1143 unquoted, rest = _unquote_c_style(name)
1144 if rest:
1145 raise ValueError(f"Trailing data after quoted name: {name!r}")
1146 return unquoted
1149def _parse_file_line_path(path: bytes) -> bytes:
1150 """Parse the path from a ``---`` or ``+++`` line, dropping any timestamp."""
1151 if not path.startswith(b'"'):
1152 return path.split(b"\t")[0]
1153 unquoted, rest = _unquote_c_style(path)
1154 if rest and not rest.startswith(b"\t"):
1155 raise ValueError(f"Trailing data after quoted name: {path!r}")
1156 return unquoted
1159def _parse_git_diff_header_paths(line: bytes) -> tuple[bytes | None, bytes | None]:
1160 """Extract the old and new paths from a ``diff --git`` line.
1162 Unquoted names are only separable when they are the same apart from their
1163 prefixes (as for anything but a rename or copy), in which case the line is
1164 split in the middle. Returns ``(None, None)`` if the names differ.
1165 """
1166 names = line[len(b"diff --git ") :]
1167 if names.startswith(b'"'):
1168 old, rest = _unquote_c_style(names)
1169 if not rest.startswith(b" "):
1170 return None, None
1171 new = _unquote_name(rest[1:])
1172 else:
1173 half = len(names) // 2
1174 if len(names) % 2 != 1 or names[half : half + 1] != b" ":
1175 return None, None
1176 old, new = names[:half], names[half + 1 :]
1177 if old.partition(b"/")[2] != new.partition(b"/")[2]:
1178 return None, None
1179 return old, new
1182def parse_unified_diff(diff_text: bytes) -> list[FilePatch]:
1183 """Parse a unified diff into FilePatch objects.
1185 Args:
1186 diff_text: Unified diff content as bytes
1188 Returns:
1189 List of FilePatch objects
1190 """
1191 patches: list[FilePatch] = []
1192 lines = diff_text.split(b"\n")
1193 i = 0
1195 while i < len(lines):
1196 line = lines[i]
1198 # Look for diff header
1199 if line.startswith(b"diff --git "):
1200 # Parse file patch
1201 old_path = None
1202 new_path = None
1203 old_mode = None
1204 new_mode = None
1205 hunks: list[PatchHunk] = []
1206 binary = False
1207 rename_from = None
1208 rename_to = None
1209 copy_from = None
1210 copy_to = None
1211 binary_old = None
1212 binary_new = None
1213 binary_old_delta = False
1214 binary_new_delta = False
1215 header_old_path, header_new_path = _parse_git_diff_header_paths(line)
1217 # Parse extended headers
1218 i += 1
1219 while i < len(lines):
1220 line = lines[i]
1222 if line.startswith(b"old file mode "):
1223 old_mode = int(line.split()[-1], 8)
1224 i += 1
1225 elif line.startswith(b"new file mode "):
1226 new_mode = int(line.split()[-1], 8)
1227 header_old_path = None
1228 i += 1
1229 elif line.startswith(b"deleted file mode "):
1230 old_mode = int(line.split()[-1], 8)
1231 header_new_path = None
1232 i += 1
1233 elif line.startswith(b"new mode "):
1234 new_mode = int(line.split()[-1], 8)
1235 i += 1
1236 elif line.startswith(b"old mode "):
1237 old_mode = int(line.split()[-1], 8)
1238 i += 1
1239 elif line.startswith(b"rename from "):
1240 rename_from = _unquote_name(line[12:].strip())
1241 i += 1
1242 elif line.startswith(b"rename to "):
1243 rename_to = _unquote_name(line[10:].strip())
1244 i += 1
1245 elif line.startswith(b"copy from "):
1246 copy_from = _unquote_name(line[10:].strip())
1247 i += 1
1248 elif line.startswith(b"copy to "):
1249 copy_to = _unquote_name(line[8:].strip())
1250 i += 1
1251 elif line.startswith(b"similarity index "):
1252 # Just skip similarity index for now
1253 i += 1
1254 elif line.startswith(b"dissimilarity index "):
1255 # Just skip dissimilarity index for now
1256 i += 1
1257 elif line.startswith(b"index "):
1258 # "index <old>..<new> <mode>" gives the mode when the
1259 # patch doesn't change it.
1260 fields = line.split()
1261 if len(fields) == 3:
1262 mode = int(fields[2], 8)
1263 if old_mode is None:
1264 old_mode = mode
1265 if new_mode is None:
1266 new_mode = mode
1267 i += 1
1268 elif line.startswith(b"--- "):
1269 # Parse old file path
1270 path = _parse_file_line_path(line[4:])
1271 if path != b"/dev/null":
1272 old_path = path
1273 i += 1
1274 elif line.startswith(b"+++ "):
1275 # Parse new file path
1276 path = _parse_file_line_path(line[4:])
1277 if path != b"/dev/null":
1278 new_path = path
1279 i += 1
1280 break
1281 elif line.startswith(b"Binary files"):
1282 binary = True
1283 i += 1
1284 break
1285 elif line.startswith(b"GIT binary patch"):
1286 binary = True
1287 i += 1
1288 # The forward data comes first, followed by the reverse
1289 # data. Each is a "literal" or "delta" line, then base85
1290 # lines up to a blank line.
1291 blocks: list[tuple[bool, bytes]] = []
1292 while (
1293 i < len(lines)
1294 and len(blocks) < 2
1295 and lines[i].startswith((b"literal ", b"delta "))
1296 ):
1297 is_delta = lines[i].startswith(b"delta ")
1298 i += 1
1299 block_start = i
1300 while i < len(lines) and lines[i]:
1301 i += 1
1302 blocks.append(
1303 (
1304 is_delta,
1305 b"".join(data + b"\n" for data in lines[block_start:i]),
1306 )
1307 )
1308 i += 1
1309 if not blocks:
1310 raise ValueError("GIT binary patch without data")
1311 binary_new_delta, binary_new = blocks[0]
1312 if len(blocks) > 1:
1313 binary_old_delta, binary_old = blocks[1]
1314 break
1315 else:
1316 # Leave the line (e.g. the next "diff --git") to the
1317 # hunk parser below.
1318 break
1320 if old_path is None and new_path is None:
1321 # Binary patches and those without content changes (mode
1322 # changes, empty files) have no ---/+++ lines.
1323 old_path = header_old_path
1324 new_path = header_new_path
1326 # Parse hunks
1327 if not binary:
1328 while i < len(lines):
1329 line = lines[i]
1331 if line.startswith(b"@@ "):
1332 # Parse hunk header
1333 match = re.match(
1334 rb"@@ -(\d+)(?:,(\d+))? \+(\d+)(?:,(\d+))? @@", line
1335 )
1336 if match:
1337 old_start = int(match.group(1))
1338 old_count = int(match.group(2)) if match.group(2) else 1
1339 new_start = int(match.group(3))
1340 new_count = int(match.group(4)) if match.group(4) else 1
1342 # Parse hunk lines
1343 hunk_lines: list[bytes] = []
1344 i += 1
1345 while i < len(lines):
1346 line = lines[i]
1347 if line.startswith((b" ", b"+", b"-", b"\\")):
1348 hunk_lines.append(line)
1349 i += 1
1350 else:
1351 break
1353 hunks.append(
1354 PatchHunk(
1355 old_start=old_start,
1356 old_count=old_count,
1357 new_start=new_start,
1358 new_count=new_count,
1359 lines=hunk_lines,
1360 )
1361 )
1362 else:
1363 i += 1
1364 elif line.startswith(b"diff --git "):
1365 # Next file patch
1366 break
1367 else:
1368 i += 1
1369 if not line.strip():
1370 # Empty line, might be end of patch or separator
1371 break
1373 patches.append(
1374 FilePatch(
1375 old_path=old_path,
1376 new_path=new_path,
1377 old_mode=old_mode,
1378 new_mode=new_mode,
1379 hunks=hunks,
1380 binary=binary,
1381 rename_from=rename_from,
1382 rename_to=rename_to,
1383 copy_from=copy_from,
1384 copy_to=copy_to,
1385 binary_old=binary_old,
1386 binary_new=binary_new,
1387 binary_old_delta=binary_old_delta,
1388 binary_new_delta=binary_new_delta,
1389 )
1390 )
1391 else:
1392 i += 1
1394 return patches
1397def apply_patch_hunks(
1398 patch: FilePatch,
1399 original_lines: list[bytes],
1400) -> list[bytes] | None:
1401 """Apply patch hunks to file content.
1403 Args:
1404 patch: FilePatch object to apply
1405 original_lines: Original file content as list of lines
1407 Returns:
1408 Patched file content as list of lines, or None if patch cannot be applied
1409 """
1410 result = original_lines[:]
1411 offset = 0 # Track line offset as we apply hunks
1413 for hunk in patch.hunks:
1414 # old_start is 1-indexed, except that a hunk without old lines
1415 # inserts after line old_start.
1416 if hunk.old_count == 0:
1417 target_line = hunk.old_start + offset
1418 else:
1419 target_line = hunk.old_start - 1 + offset
1421 # Extract old and new content from hunk
1422 old_content: list[bytes] = []
1423 new_content: list[bytes] = []
1425 previous = b""
1426 for line in hunk.lines:
1427 if line.startswith(b"\\"):
1428 # "\ No newline at end of file" applies to the line before it
1429 if previous in (b" ", b"-"):
1430 old_content[-1] = old_content[-1].removesuffix(b"\n")
1431 if previous in (b" ", b"+"):
1432 new_content[-1] = new_content[-1].removesuffix(b"\n")
1433 continue
1434 previous = line[:1]
1435 if line.startswith(b" "):
1436 # Context line - add newline if not present
1437 content = line[1:]
1438 if not content.endswith(b"\n"):
1439 content += b"\n"
1440 old_content.append(content)
1441 new_content.append(content)
1442 elif line.startswith(b"-"):
1443 # Deletion - add newline if not present
1444 content = line[1:]
1445 if not content.endswith(b"\n"):
1446 content += b"\n"
1447 old_content.append(content)
1448 elif line.startswith(b"+"):
1449 # Addition - add newline if not present
1450 content = line[1:]
1451 if not content.endswith(b"\n"):
1452 content += b"\n"
1453 new_content.append(content)
1455 # Verify context matches
1456 if target_line < 0 or target_line + len(old_content) > len(result):
1457 # TODO: Implement fuzzy matching
1458 return None
1460 for i, old_line in enumerate(old_content):
1461 if result[target_line + i] != old_line:
1462 # Context doesn't match
1463 # TODO: Implement fuzzy matching
1464 return None
1466 # Apply the patch
1467 result[target_line : target_line + len(old_content)] = new_content
1469 # Update offset for next hunk
1470 offset += len(new_content) - len(old_content)
1472 return result
1475def _ensure_within_repo(repo_path: bytes, fs_path: bytes, rel_path: bytes) -> None:
1476 """Reject patch target paths that resolve outside the working tree.
1478 Patch headers are untrusted (e.g. ``git am`` of a mailbox), so a name such
1479 as ``../../etc/cron.d/x`` or an absolute path must not be written through
1480 ``os.path.join``. Mirrors git's refusal of paths outside the work tree.
1481 """
1482 repo_root = os.path.realpath(repo_path)
1483 resolved = os.path.realpath(fs_path)
1484 if resolved != repo_root and not resolved.startswith(
1485 repo_root + os.fsencode(os.sep)
1486 ):
1487 raise ValueError(f"patch affects file outside repository: {rel_path!r}")
1490def _validate_patch_target(r: "Repo", repo_path: bytes, tree_path: bytes) -> bytes:
1491 """Validate a patch target path and return its filesystem path.
1493 ``_ensure_within_repo`` alone only refuses paths that escape the work tree;
1494 a target like ``.git/hooks/pre-commit`` stays inside it and would otherwise
1495 be written (and later executed as a hook). On top of that containment check,
1496 apply the same name and symlink checks git uses for checkout, so patch
1497 application cannot reach the control directory. Mirrors
1498 ``porcelain._checked_worktree_path``.
1500 Returns:
1501 The filesystem path under ``repo_path``, as bytes.
1502 """
1503 from .index import (
1504 InvalidPathError,
1505 get_path_element_validator,
1506 validate_path,
1507 verify_leading_dirs,
1508 )
1510 fs_path = os.path.join(repo_path, tree_path)
1511 _ensure_within_repo(repo_path, fs_path, tree_path)
1512 validator = get_path_element_validator(r.get_config_stack())
1513 if not validate_path(tree_path, validator):
1514 raise ValueError(f"refusing to write unsafe path: {tree_path!r}")
1515 try:
1516 verify_leading_dirs(tree_path, [], repo_path)
1517 except InvalidPathError:
1518 raise ValueError(f"refusing to write through symlink: {tree_path!r}")
1519 return fs_path
1522def _read_patch_target(fs_path: bytes) -> tuple[bytes, int] | None:
1523 """Read the content and mode of a patch target in the work tree.
1525 A symlink is read with ``readlink`` rather than followed, since its
1526 content in git is the link target.
1528 Returns:
1529 Tuple with content and mode, or None if the path does not exist
1530 """
1531 from .index import cleanup_mode
1533 try:
1534 st = os.lstat(fs_path)
1535 except FileNotFoundError:
1536 return None
1537 if stat.S_ISLNK(st.st_mode):
1538 return os.readlink(fs_path), stat.S_IFLNK
1539 with open(fs_path, "rb") as f:
1540 return f.read(), cleanup_mode(st.st_mode)
1543def _check_patch_target_type(
1544 tree_path: bytes, disk_mode: int, expected_mode: int, config: "Config"
1545) -> None:
1546 """Refuse to apply a patch for a symlink to a file or vice versa.
1548 ``verify_leading_dirs`` only checks the leading directories. A tracked
1549 symlink such as ``trap -> .git/hooks/pre-commit`` resolves inside the work
1550 tree, so writing a regular file patch to it would follow the link and land
1551 in the control directory. git refuses such patches with "wrong type".
1552 """
1553 if stat.S_ISLNK(disk_mode) == stat.S_ISLNK(expected_mode):
1554 return
1555 if (
1556 stat.S_ISLNK(expected_mode)
1557 and not stat.S_ISLNK(disk_mode)
1558 and not config.get_boolean(b"core", b"symlinks", True)
1559 ):
1560 # With core.symlinks=false links are checked out as plain files.
1561 return
1562 raise ValueError(f"wrong type for patch target: {tree_path!r}")
1565def _write_patch_target(
1566 fs_path: bytes, content: bytes, mode: int, config: "Config"
1567) -> os.stat_result:
1568 """Write a patch result to the work tree without following symlinks.
1570 Returns:
1571 The ``lstat`` result of the written path
1572 """
1573 from .index import build_file_from_blob, get_symlink_fn
1575 os.makedirs(os.path.dirname(fs_path), exist_ok=True)
1576 return build_file_from_blob(
1577 Blob.from_string(content), mode, fs_path, symlink_fn=get_symlink_fn(config)
1578 )
1581def _refuse_existing_target(fs_path: bytes, tree_path: bytes) -> None:
1582 """Refuse to create a file over an existing path, like git does."""
1583 if os.path.lexists(fs_path):
1584 raise ValueError(f"{tree_path!r} already exists in working directory")
1587def _load_patch_target(
1588 r: "Repo",
1589 fs_path: bytes,
1590 tree_path: bytes,
1591 expected_mode: int | None,
1592 config: "Config",
1593) -> tuple[bytes, int] | None:
1594 """Load the current content and mode of an existing patch target.
1596 Reads the work tree, falling back to the index if the path is missing
1597 there. The mode from the patch, if any, must match the type on disk.
1599 Returns:
1600 Tuple with content and mode, or None if the path can't be found
1601 """
1602 from .index import ConflictedIndexEntry, IndexEntry
1604 def index_entry() -> IndexEntry | None:
1605 try:
1606 entry = r.open_index(config=config)[tree_path]
1607 except (FileNotFoundError, KeyError):
1608 return None
1609 if isinstance(entry, ConflictedIndexEntry):
1610 return None
1611 return entry
1613 current = _read_patch_target(fs_path)
1614 if current is None:
1615 entry = index_entry()
1616 if entry is None:
1617 return None
1618 obj = r.object_store[entry.sha]
1619 if not isinstance(obj, Blob):
1620 return None
1621 return obj.data, entry.mode
1622 content, mode = current
1623 if expected_mode is not None:
1624 _check_patch_target_type(tree_path, mode, expected_mode, config)
1625 return content, expected_mode
1626 if not stat.S_ISLNK(mode) and not config.get_boolean(b"core", b"symlinks", True):
1627 # A link checked out as a plain file is only recognizable as such
1628 # from the index.
1629 entry = index_entry()
1630 if entry is not None and stat.S_ISLNK(entry.mode):
1631 return content, entry.mode
1632 return content, mode
1635def _apply_rename_or_copy(
1636 r: "Repo",
1637 src_path: bytes,
1638 dst_path: bytes,
1639 strip: int,
1640 patch: FilePatch,
1641 is_rename: bool,
1642 cached: bool,
1643 check: bool,
1644 config: "Config",
1645) -> tuple[list[bytes] | None, int, bool]:
1646 """Apply a rename or copy operation.
1648 Args:
1649 r: Repository object
1650 src_path: Source path
1651 dst_path: Destination path
1652 strip: Number of path components to strip
1653 patch: FilePatch object
1654 is_rename: True for rename, False for copy
1655 cached: Apply to index only, not working tree
1656 check: Check only, don't apply
1657 config: Repository configuration
1659 Returns:
1660 A tuple of (``original_lines``, ``old_mode``, ``should_continue``) where:
1661 - ``original_lines``: Content lines if hunks need to be applied, None otherwise
1662 - ``old_mode``: Mode of the source
1663 - ``should_continue``: True to skip to next patch, False to continue processing
1664 """
1665 from .index import IndexEntry, index_entry_from_stat
1667 # Strip path components
1668 src_stripped = src_path
1669 dst_stripped = dst_path
1670 if strip > 0:
1671 src_parts = src_path.split(b"/")
1672 if len(src_parts) > strip:
1673 src_stripped = b"/".join(src_parts[strip:])
1674 dst_parts = dst_path.split(b"/")
1675 if len(dst_parts) > strip:
1676 dst_stripped = b"/".join(dst_parts[strip:])
1678 repo_path_bytes = r.path.encode("utf-8") if isinstance(r.path, str) else r.path
1679 src_fs_path = _validate_patch_target(r, repo_path_bytes, src_stripped)
1680 dst_fs_path = _validate_patch_target(r, repo_path_bytes, dst_stripped)
1681 if not cached:
1682 _refuse_existing_target(dst_fs_path, dst_stripped)
1684 # Read content from source file
1685 op_name = "rename" if is_rename else "copy"
1686 current = _load_patch_target(r, src_fs_path, src_stripped, patch.old_mode, config)
1687 if current is None:
1688 raise ValueError(
1689 f"Cannot {op_name}: source {src_stripped.decode('utf-8', errors='replace')} not found"
1690 )
1691 content, old_mode = current
1693 # If the content changes too, return it for further processing
1694 if patch.hunks or patch.binary:
1695 return content.splitlines(keepends=True), old_mode, False
1697 # No hunks - pure rename/copy
1698 if check:
1699 return None, old_mode, True
1701 new_mode = patch.new_mode or old_mode
1702 index = r.open_index(config=config)
1703 blob = Blob.from_string(content)
1704 r.object_store.add_object(blob)
1706 if not cached:
1707 st = _write_patch_target(dst_fs_path, content, new_mode, config)
1708 entry = index_entry_from_stat(st, blob.id, mode=new_mode)
1709 else:
1710 entry = IndexEntry(
1711 ctime=(0, 0),
1712 mtime=(0, 0),
1713 dev=0,
1714 ino=0,
1715 mode=new_mode,
1716 uid=0,
1717 gid=0,
1718 size=len(content),
1719 sha=blob.id,
1720 flags=0,
1721 )
1723 index[dst_stripped] = entry
1725 # For renames, remove the old file
1726 if is_rename:
1727 if not cached and os.path.lexists(src_fs_path):
1728 os.remove(src_fs_path)
1729 if src_stripped in index:
1730 del index[src_stripped]
1732 index.write()
1733 return None, old_mode, True
1736def _apply_binary_patch(patch: FilePatch, original: bytes, reverse: bool) -> bytes:
1737 """Compute the new content of a file from a ``GIT binary patch``.
1739 Args:
1740 patch: Binary FilePatch
1741 original: Current content of the file
1742 reverse: Apply the reverse data instead of the forward data
1744 Returns:
1745 The patched content
1747 Raises:
1748 ValueError: If the patch data is invalid or doesn't apply
1749 """
1750 from .errors import ApplyDeltaError
1751 from .pack import apply_delta
1753 if reverse:
1754 data, is_delta = patch.binary_old, patch.binary_old_delta
1755 else:
1756 data, is_delta = patch.binary_new, patch.binary_new_delta
1757 if data is None:
1758 # "Binary files differ" message without actual patch data
1759 raise NotImplementedError(
1760 "Binary patch detected but no patch data provided (use git diff --binary)"
1761 )
1762 try:
1763 decoded = zlib.decompress(git_base85_decode(data))
1764 except (ValueError, zlib.error) as e:
1765 raise ValueError(f"Failed to decode binary patch: {e}")
1766 if not is_delta:
1767 return decoded
1768 try:
1769 return b"".join(apply_delta(original, decoded))
1770 except ApplyDeltaError as e:
1771 raise ValueError(f"Binary delta does not apply: {e}")
1774def apply_patches(
1775 r: "Repo",
1776 patches: list[FilePatch],
1777 cached: bool = False,
1778 reverse: bool = False,
1779 check: bool = False,
1780 strip: int = 1,
1781 three_way: bool = False,
1782 *,
1783 config: "Config | None" = None,
1784) -> None:
1785 """Apply a list of file patches to a repository.
1787 Args:
1788 r: Repository object
1789 patches: List of FilePatch objects to apply
1790 cached: Apply patch to index only, not working tree
1791 reverse: Apply patch in reverse
1792 check: Only check if patch can be applied, don't apply
1793 strip: Number of leading path components to strip (default: 1)
1794 three_way: Fall back to 3-way merge if patch does not apply cleanly
1795 config: Repository configuration. If None, falls back to
1796 ``r.get_config_stack()``.
1798 Raises:
1799 ValueError: If patch cannot be applied
1800 """
1801 from .index import (
1802 IndexEntry,
1803 index_entry_from_stat,
1804 )
1806 if config is None:
1807 config = r.get_config_stack()
1809 for patch in patches:
1810 # Determine the file path
1811 # For renames/copies without hunks, old_path/new_path may be None
1812 # Use local variables to avoid mutating the patch object
1813 old_path = patch.old_path
1814 new_path = patch.new_path
1816 if new_path is None and old_path is None:
1817 if patch.rename_to is not None:
1818 # Use rename_to for the target path
1819 new_path = patch.rename_to
1820 old_path = patch.rename_from
1821 elif patch.copy_to is not None:
1822 # Use copy_to for the target path
1823 new_path = patch.copy_to
1824 old_path = patch.copy_from
1825 else:
1826 raise ValueError("Patch has no file path")
1828 # Choose path based on operation
1829 file_path: bytes
1830 if new_path is None:
1831 # Deletion
1832 if old_path is None:
1833 raise ValueError("Patch has no file path")
1834 file_path = old_path
1835 elif old_path is None:
1836 # Addition
1837 file_path = new_path
1838 else:
1839 # Modification (use new path)
1840 file_path = new_path
1842 # Strip path components
1843 if strip > 0:
1844 parts = file_path.split(b"/")
1845 if len(parts) > strip:
1846 file_path = b"/".join(parts[strip:])
1848 # Convert to filesystem path
1849 tree_path = file_path
1850 repo_path_bytes = r.path.encode("utf-8") if isinstance(r.path, str) else r.path
1851 fs_path = _validate_patch_target(r, repo_path_bytes, file_path)
1853 # Handle renames and copies
1854 original_lines: list[bytes] | None = None
1855 old_mode: int | None = None
1856 if patch.rename_from is not None and patch.rename_to is not None:
1857 original_lines, old_mode, should_continue = _apply_rename_or_copy(
1858 r,
1859 patch.rename_from,
1860 patch.rename_to,
1861 strip,
1862 patch,
1863 is_rename=True,
1864 cached=cached,
1865 check=check,
1866 config=config,
1867 )
1868 if should_continue:
1869 continue
1870 elif patch.copy_from is not None and patch.copy_to is not None:
1871 original_lines, old_mode, should_continue = _apply_rename_or_copy(
1872 r,
1873 patch.copy_from,
1874 patch.copy_to,
1875 strip,
1876 patch,
1877 is_rename=False,
1878 cached=cached,
1879 check=check,
1880 config=config,
1881 )
1882 if should_continue:
1883 continue
1885 # Read original file content (unless already loaded from rename/copy)
1886 if original_lines is None:
1887 if old_path is None:
1888 # New file
1889 if not cached:
1890 _refuse_existing_target(fs_path, tree_path)
1891 original_lines = []
1892 else:
1893 current = _load_patch_target(
1894 r, fs_path, tree_path, patch.old_mode, config
1895 )
1896 if current is None:
1897 original_lines = []
1898 else:
1899 content, old_mode = current
1900 original_lines = content.splitlines(keepends=True)
1901 new_mode = patch.new_mode or old_mode or patch.old_mode or 0o100644
1903 # Reverse patch if requested
1904 if reverse:
1905 # Swap old and new in hunks
1906 for hunk in patch.hunks:
1907 hunk.old_start, hunk.new_start = hunk.new_start, hunk.old_start
1908 hunk.old_count, hunk.new_count = hunk.new_count, hunk.old_count
1909 # Swap +/- prefixes
1910 reversed_lines = []
1911 for line in hunk.lines:
1912 if line.startswith(b"+"):
1913 reversed_lines.append(b"-" + line[1:])
1914 elif line.startswith(b"-"):
1915 reversed_lines.append(b"+" + line[1:])
1916 else:
1917 reversed_lines.append(line)
1918 hunk.lines = reversed_lines
1920 # Apply the patch
1921 assert original_lines is not None
1922 result: list[bytes] | None
1923 # Deletions don't need the binary data, so treat those like text.
1924 if patch.binary and new_path is not None:
1925 result = _apply_binary_patch(
1926 patch, b"".join(original_lines), reverse
1927 ).splitlines(keepends=True)
1928 else:
1929 result = apply_patch_hunks(patch, original_lines)
1931 if result is None and three_way:
1932 # Try 3-way merge fallback
1933 from .merge import merge_blobs
1935 # Reconstruct base version from the patch
1936 # Base is what you get by taking only the old lines from hunks
1937 base_lines = []
1938 theirs_lines = []
1940 for hunk in patch.hunks:
1941 for line in hunk.lines:
1942 if line.startswith(b"\\"):
1943 # Skip "\ No newline at end of file" markers
1944 continue
1945 elif line.startswith(b" "):
1946 # Context line - in both base and theirs
1947 content = line[1:]
1948 if not content.endswith(b"\n"):
1949 content += b"\n"
1950 base_lines.append(content)
1951 theirs_lines.append(content)
1952 elif line.startswith(b"-"):
1953 # Deletion - only in base
1954 content = line[1:]
1955 if not content.endswith(b"\n"):
1956 content += b"\n"
1957 base_lines.append(content)
1958 elif line.startswith(b"+"):
1959 # Addition - only in theirs
1960 content = line[1:]
1961 if not content.endswith(b"\n"):
1962 content += b"\n"
1963 theirs_lines.append(content)
1965 # Create blobs for merging
1966 base_content = b"".join(base_lines)
1967 ours_content = b"".join(original_lines)
1968 theirs_content = b"".join(theirs_lines)
1970 base_blob = Blob.from_string(base_content) if base_content else None
1971 ours_blob = Blob.from_string(ours_content) if ours_content else None
1972 theirs_blob = Blob.from_string(theirs_content)
1974 # Perform 3-way merge
1975 merged_content, _had_conflicts = merge_blobs(
1976 base_blob, ours_blob, theirs_blob, path=tree_path
1977 )
1979 result = merged_content.splitlines(keepends=True)
1981 # Note: if _had_conflicts is True, the result contains conflict markers
1982 # Git would exit with error code, but we continue processing
1983 elif result is None:
1984 raise PatchApplicationFailure(
1985 f"Patch does not apply to {file_path.decode('utf-8', errors='replace')}"
1986 )
1988 if check:
1989 # Just checking, don't actually apply
1990 continue
1992 # Write result
1993 result_content = b"".join(result)
1995 if new_path is None:
1996 # File deletion
1997 if not cached and os.path.lexists(fs_path):
1998 os.remove(fs_path)
1999 # Remove from index
2000 index = r.open_index(config=config)
2001 if tree_path in index:
2002 del index[tree_path]
2003 index.write()
2004 else:
2005 # File addition or modification
2006 index = r.open_index(config=config)
2007 blob = Blob.from_string(result_content)
2008 r.object_store.add_object(blob)
2010 if not cached:
2011 st = _write_patch_target(fs_path, result_content, new_mode, config)
2012 entry = index_entry_from_stat(st, blob.id, mode=new_mode)
2013 else:
2014 # Create a minimal index entry for cached-only changes
2015 entry = IndexEntry(
2016 ctime=(0, 0),
2017 mtime=(0, 0),
2018 dev=0,
2019 ino=0,
2020 mode=new_mode,
2021 uid=0,
2022 gid=0,
2023 size=len(result_content),
2024 sha=blob.id,
2025 flags=0,
2026 )
2028 index[tree_path] = entry
2030 # Handle cleanup for renames with hunks
2031 if patch.rename_from is not None and patch.rename_to is not None:
2032 # Remove old file after successful rename
2033 old_rename_path = patch.rename_from
2034 if strip > 0:
2035 old_parts = old_rename_path.split(b"/")
2036 if len(old_parts) > strip:
2037 old_rename_path = b"/".join(old_parts[strip:])
2039 old_fs_path = os.path.join(
2040 r.path.encode("utf-8") if isinstance(r.path, str) else r.path,
2041 old_rename_path,
2042 )
2044 if not cached and os.path.lexists(old_fs_path):
2045 os.remove(old_fs_path)
2046 if old_rename_path in index:
2047 del index[old_rename_path]
2049 index.write()
2052def mailinfo(
2053 msg: email.message.Message | BinaryIO | TextIO,
2054 keep_subject: bool = False,
2055 keep_non_patch: bool = False,
2056 encoding: str | None = None,
2057 scissors: bool = False,
2058 message_id: bool = False,
2059) -> MailinfoResult:
2060 """Extract patch information from an email message.
2062 This function parses an email message and extracts commit metadata
2063 (author, email, subject) and separates the commit message from the
2064 patch content, similar to git mailinfo.
2066 Args:
2067 msg: Email message (email.message.Message object) or file handle to read from
2068 keep_subject: If True, keep subject intact without munging (-k)
2069 keep_non_patch: If True, only strip [PATCH] from brackets (-b)
2070 encoding: Character encoding to use (default: detect from message)
2071 scissors: If True, remove everything before scissors line
2072 message_id: If True, include Message-ID in commit message (-m)
2074 Returns:
2075 MailinfoResult with parsed information
2077 Raises:
2078 ValueError: If message is malformed or missing required fields
2079 """
2080 # Parse message if given a file handle
2081 parsed_msg: email.message.Message
2082 if not isinstance(msg, email.message.Message):
2083 if hasattr(msg, "read"):
2084 content = msg.read()
2085 if isinstance(content, bytes):
2086 bparser = email.parser.BytesParser()
2087 parsed_msg = bparser.parsebytes(content)
2088 else:
2089 sparser = email.parser.Parser()
2090 parsed_msg = sparser.parsestr(content)
2091 else:
2092 raise ValueError("msg must be an email.message.Message or file-like object")
2093 else:
2094 parsed_msg = msg
2096 # Detect encoding from message if not specified
2097 if encoding is None:
2098 encoding = parsed_msg.get_content_charset() or "utf-8"
2100 # Extract author information
2101 from_header = parsed_msg.get("From", "")
2102 if not from_header:
2103 raise ValueError("Email message missing 'From' header")
2105 # Parse "Name <email>" format
2106 author_name, author_email = email.utils.parseaddr(from_header)
2107 if not author_email:
2108 raise ValueError(
2109 f"Could not parse email address from 'From' header: {from_header}"
2110 )
2112 # Extract date
2113 date_header = parsed_msg.get("Date")
2114 author_date = date_header if date_header else None
2116 # Extract and process subject
2117 subject = parsed_msg.get("Subject", "")
2118 if not subject:
2119 subject = "(no subject)"
2121 # Convert Header object to string if needed
2122 subject = str(subject)
2124 # Remove newlines from subject
2125 subject = subject.replace("\n", " ").replace("\r", " ")
2126 subject = _munge_subject(subject, keep_subject, keep_non_patch)
2128 # Extract Message-ID if requested
2129 msg_id = None
2130 if message_id:
2131 msg_id = parsed_msg.get("Message-ID")
2133 # Get message body
2134 body = parsed_msg.get_payload(decode=True)
2135 if body is None:
2136 body = b""
2137 elif isinstance(body, str):
2138 body = body.encode(encoding)
2139 elif not isinstance(body, bytes):
2140 # Handle multipart or other types
2141 body = str(body).encode(encoding)
2143 # Split into lines
2144 lines = body.splitlines(keepends=True)
2146 # Handle scissors
2147 scissors_idx = None
2148 if scissors:
2149 scissors_idx = _find_scissors_line(lines)
2150 if scissors_idx is not None:
2151 # Remove everything up to and including scissors line
2152 lines = lines[scissors_idx + 1 :]
2154 # Separate commit message from patch
2155 # Look for the "---" separator that indicates start of diffstat/patch
2156 message_lines: list[bytes] = []
2157 patch_lines: list[bytes] = []
2158 in_patch = False
2160 for line in lines:
2161 if not in_patch and line == b"---\n":
2162 in_patch = True
2163 patch_lines.append(line)
2164 elif in_patch:
2165 # Stop at signature marker "-- "
2166 if line == b"-- \n":
2167 break
2168 patch_lines.append(line)
2169 else:
2170 message_lines.append(line)
2172 # Build commit message
2173 commit_message = b"".join(message_lines).decode(encoding, errors="replace")
2175 # Clean up commit message
2176 commit_message = commit_message.strip()
2178 # Append Message-ID if requested
2179 if message_id and msg_id:
2180 if commit_message:
2181 commit_message += "\n\n"
2182 commit_message += f"Message-ID: {msg_id}"
2184 # Build patch content
2185 patch_content = b"".join(patch_lines).decode(encoding, errors="replace")
2187 return MailinfoResult(
2188 author_name=author_name,
2189 author_email=author_email,
2190 author_date=author_date,
2191 subject=subject,
2192 message=commit_message,
2193 patch=patch_content,
2194 message_id=msg_id,
2195 )