Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/dulwich/config.py: 61%
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# config.py - Reading and writing Git config files
2# Copyright (C) 2011-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"""Reading and writing Git configuration files.
24Todo:
25 * preserve formatting when updating configuration files
26"""
28import logging
29import os
30import re
31import sys
32from collections.abc import (
33 ItemsView,
34 Iterable,
35 Iterator,
36 KeysView,
37 MutableMapping,
38 ValuesView,
39)
40from contextlib import suppress
41from pathlib import Path
42from typing import (
43 IO,
44 Callable,
45 Generic,
46 Optional,
47 TypeVar,
48 Union,
49 overload,
50)
52from .file import GitFile, _GitFile
54ConfigKey = Union[str, bytes, tuple[Union[str, bytes], ...]]
55ConfigValue = Union[str, bytes, bool, int]
57logger = logging.getLogger(__name__)
59# Type for file opener callback
60FileOpener = Callable[[Union[str, os.PathLike]], IO[bytes]]
62# Type for includeIf condition matcher
63# Takes the condition value (e.g., "main" for onbranch:main) and returns bool
64ConditionMatcher = Callable[[str], bool]
66# Security limits for include files
67MAX_INCLUDE_FILE_SIZE = 1024 * 1024 # 1MB max for included config files
68DEFAULT_MAX_INCLUDE_DEPTH = 10 # Maximum recursion depth for includes
71def _match_gitdir_pattern(
72 path: bytes, pattern: bytes, ignorecase: bool = False
73) -> bool:
74 """Simple gitdir pattern matching for includeIf conditions.
76 This handles the basic gitdir patterns used in includeIf directives.
77 """
78 # Convert to strings for easier manipulation
79 path_str = path.decode("utf-8", errors="replace")
80 pattern_str = pattern.decode("utf-8", errors="replace")
82 # Normalize paths to use forward slashes for consistent matching
83 path_str = path_str.replace("\\", "/")
84 pattern_str = pattern_str.replace("\\", "/")
86 if ignorecase:
87 path_str = path_str.lower()
88 pattern_str = pattern_str.lower()
90 # Handle the common cases for gitdir patterns
91 if pattern_str.startswith("**/") and pattern_str.endswith("/**"):
92 # Pattern like **/dirname/** should match any path containing dirname
93 dirname = pattern_str[3:-3] # Remove **/ and /**
94 # Check if path contains the directory name as a path component
95 return ("/" + dirname + "/") in path_str or path_str.endswith("/" + dirname)
96 elif pattern_str.startswith("**/"):
97 # Pattern like **/filename
98 suffix = pattern_str[3:] # Remove **/
99 return suffix in path_str or path_str.endswith("/" + suffix)
100 elif pattern_str.endswith("/**"):
101 # Pattern like /path/to/dir/** should match /path/to/dir and any subdirectory
102 base_pattern = pattern_str[:-3] # Remove /**
103 return path_str == base_pattern or path_str.startswith(base_pattern + "/")
104 elif "**" in pattern_str:
105 # Handle patterns with ** in the middle
106 parts = pattern_str.split("**")
107 if len(parts) == 2:
108 prefix, suffix = parts
109 # Path must start with prefix and end with suffix (if any)
110 if prefix and not path_str.startswith(prefix):
111 return False
112 if suffix and not path_str.endswith(suffix):
113 return False
114 return True
116 # Direct match or simple glob pattern
117 if "*" in pattern_str or "?" in pattern_str or "[" in pattern_str:
118 import fnmatch
120 return fnmatch.fnmatch(path_str, pattern_str)
121 else:
122 return path_str == pattern_str
125def match_glob_pattern(value: str, pattern: str) -> bool:
126 r"""Match a value against a glob pattern.
128 Supports simple glob patterns like ``*`` and ``**``.
130 Raises:
131 ValueError: If the pattern is invalid
132 """
133 # Convert glob pattern to regex
134 pattern_escaped = re.escape(pattern)
135 # Replace escaped \*\* with .* (match anything)
136 pattern_escaped = pattern_escaped.replace(r"\*\*", ".*")
137 # Replace escaped \* with [^/]* (match anything except /)
138 pattern_escaped = pattern_escaped.replace(r"\*", "[^/]*")
139 # Anchor the pattern
140 pattern_regex = f"^{pattern_escaped}$"
142 try:
143 return bool(re.match(pattern_regex, value))
144 except re.error as e:
145 raise ValueError(f"Invalid glob pattern {pattern!r}: {e}")
148def lower_key(key: ConfigKey) -> ConfigKey:
149 """Convert a config key to lowercase, preserving subsection case.
151 Args:
152 key: Configuration key (str, bytes, or tuple)
154 Returns:
155 Key with section names lowercased, subsection names preserved
157 Raises:
158 TypeError: If key is not str, bytes, or tuple
159 """
160 if isinstance(key, (bytes, str)):
161 return key.lower()
163 if isinstance(key, tuple):
164 # For config sections, only lowercase the section name (first element)
165 # but preserve the case of subsection names (remaining elements)
166 if len(key) > 0:
167 first = key[0]
168 assert isinstance(first, (bytes, str))
169 return (first.lower(), *key[1:])
170 return key
172 raise TypeError(key)
175K = TypeVar("K", bound=ConfigKey) # Key type must be ConfigKey
176V = TypeVar("V") # Value type
177_T = TypeVar("_T") # For get() default parameter
180class CaseInsensitiveOrderedMultiDict(MutableMapping[K, V], Generic[K, V]):
181 """A case-insensitive ordered dictionary that can store multiple values per key.
183 This class maintains the order of insertions and allows multiple values
184 for the same key. Keys are compared case-insensitively.
185 """
187 def __init__(self, default_factory: Optional[Callable[[], V]] = None) -> None:
188 """Initialize a CaseInsensitiveOrderedMultiDict.
190 Args:
191 default_factory: Optional factory function for default values
192 """
193 self._real: list[tuple[K, V]] = []
194 self._keyed: dict[ConfigKey, V] = {}
195 self._default_factory = default_factory
197 @classmethod
198 def make(
199 cls,
200 dict_in: Optional[
201 Union[MutableMapping[K, V], "CaseInsensitiveOrderedMultiDict[K, V]"]
202 ] = None,
203 default_factory: Optional[Callable[[], V]] = None,
204 ) -> "CaseInsensitiveOrderedMultiDict[K, V]":
205 """Create a CaseInsensitiveOrderedMultiDict from an existing mapping.
207 Args:
208 dict_in: Optional mapping to initialize from
209 default_factory: Optional factory function for default values
211 Returns:
212 New CaseInsensitiveOrderedMultiDict instance
214 Raises:
215 TypeError: If dict_in is not a mapping or None
216 """
217 if isinstance(dict_in, cls):
218 return dict_in
220 out = cls(default_factory=default_factory)
222 if dict_in is None:
223 return out
225 if not isinstance(dict_in, MutableMapping):
226 raise TypeError
228 for key, value in dict_in.items():
229 out[key] = value
231 return out
233 def __len__(self) -> int:
234 """Return the number of unique keys in the dictionary."""
235 return len(self._keyed)
237 def keys(self) -> KeysView[K]:
238 """Return a view of the dictionary's keys."""
239 # Return a view of the original keys (not lowercased)
240 # We need to deduplicate since _real can have duplicates
241 seen = set()
242 unique_keys = []
243 for k, _ in self._real:
244 lower = lower_key(k)
245 if lower not in seen:
246 seen.add(lower)
247 unique_keys.append(k)
248 from collections.abc import KeysView as ABCKeysView
250 class UniqueKeysView(ABCKeysView[K]):
251 def __init__(self, keys: list[K]):
252 self._keys = keys
254 def __contains__(self, key: object) -> bool:
255 return key in self._keys
257 def __iter__(self):
258 return iter(self._keys)
260 def __len__(self) -> int:
261 return len(self._keys)
263 return UniqueKeysView(unique_keys)
265 def items(self) -> ItemsView[K, V]:
266 """Return a view of the dictionary's (key, value) pairs in insertion order."""
268 # Return a view that iterates over the real list to preserve order
269 class OrderedItemsView(ItemsView[K, V]):
270 """Items view that preserves insertion order."""
272 def __init__(self, mapping: CaseInsensitiveOrderedMultiDict[K, V]):
273 self._mapping = mapping
275 def __iter__(self) -> Iterator[tuple[K, V]]:
276 return iter(self._mapping._real)
278 def __len__(self) -> int:
279 return len(self._mapping._real)
281 def __contains__(self, item: object) -> bool:
282 if not isinstance(item, tuple) or len(item) != 2:
283 return False
284 key, value = item
285 return any(k == key and v == value for k, v in self._mapping._real)
287 return OrderedItemsView(self)
289 def __iter__(self) -> Iterator[K]:
290 """Iterate over the dictionary's keys."""
291 # Return iterator over original keys (not lowercased), deduplicated
292 seen = set()
293 for k, _ in self._real:
294 lower = lower_key(k)
295 if lower not in seen:
296 seen.add(lower)
297 yield k
299 def values(self) -> ValuesView[V]:
300 """Return a view of the dictionary's values."""
301 return self._keyed.values()
303 def __setitem__(self, key: K, value: V) -> None:
304 """Set a value for a key, appending to existing values."""
305 self._real.append((key, value))
306 self._keyed[lower_key(key)] = value
308 def set(self, key: K, value: V) -> None:
309 """Set a value for a key, replacing all existing values.
311 Args:
312 key: The key to set
313 value: The value to set
314 """
315 # This method replaces all existing values for the key
316 lower = lower_key(key)
317 self._real = [(k, v) for k, v in self._real if lower_key(k) != lower]
318 self._real.append((key, value))
319 self._keyed[lower] = value
321 def __delitem__(self, key: K) -> None:
322 """Delete all values for a key.
324 Raises:
325 KeyError: If the key is not found
326 """
327 lower_k = lower_key(key)
328 del self._keyed[lower_k]
329 for i, (actual, unused_value) in reversed(list(enumerate(self._real))):
330 if lower_key(actual) == lower_k:
331 del self._real[i]
333 def __getitem__(self, item: K) -> V:
334 """Get the last value for a key.
336 Raises:
337 KeyError: If the key is not found
338 """
339 return self._keyed[lower_key(item)]
341 def get(self, key: K, /, default: Union[V, _T, None] = None) -> Union[V, _T, None]: # type: ignore[override]
342 """Get the last value for a key, or a default if not found.
344 Args:
345 key: The key to look up
346 default: Default value to return if key not found
348 Returns:
349 The value for the key, or default/default_factory result if not found
350 """
351 try:
352 return self[key]
353 except KeyError:
354 if default is not None:
355 return default
356 elif self._default_factory is not None:
357 return self._default_factory()
358 else:
359 return None
361 def get_all(self, key: K) -> Iterator[V]:
362 """Get all values for a key in insertion order.
364 Args:
365 key: The key to look up
367 Returns:
368 Iterator of all values for the key
369 """
370 lowered_key = lower_key(key)
371 for actual, value in self._real:
372 if lower_key(actual) == lowered_key:
373 yield value
375 def setdefault(self, key: K, default: Optional[V] = None) -> V:
376 """Get value for key, setting it to default if not present.
378 Args:
379 key: The key to look up
380 default: Default value to set if key not found
382 Returns:
383 The existing value or the newly set default
385 Raises:
386 KeyError: If key not found and no default or default_factory
387 """
388 try:
389 return self[key]
390 except KeyError:
391 if default is not None:
392 self[key] = default
393 return default
394 elif self._default_factory is not None:
395 value = self._default_factory()
396 self[key] = value
397 return value
398 else:
399 raise
402Name = bytes
403NameLike = Union[bytes, str]
404Section = tuple[bytes, ...]
405SectionLike = Union[bytes, str, tuple[Union[bytes, str], ...]]
406Value = bytes
407ValueLike = Union[bytes, str]
410class Config:
411 """A Git configuration."""
413 def get(self, section: SectionLike, name: NameLike) -> Value:
414 """Retrieve the contents of a configuration setting.
416 Args:
417 section: Tuple with section name and optional subsection name
418 name: Variable name
419 Returns:
420 Contents of the setting
421 Raises:
422 KeyError: if the value is not set
423 """
424 raise NotImplementedError(self.get)
426 def get_multivar(self, section: SectionLike, name: NameLike) -> Iterator[Value]:
427 """Retrieve the contents of a multivar configuration setting.
429 Args:
430 section: Tuple with section name and optional subsection namee
431 name: Variable name
432 Returns:
433 Contents of the setting as iterable
434 Raises:
435 KeyError: if the value is not set
436 """
437 raise NotImplementedError(self.get_multivar)
439 @overload
440 def get_boolean(
441 self, section: SectionLike, name: NameLike, default: bool
442 ) -> bool: ...
444 @overload
445 def get_boolean(self, section: SectionLike, name: NameLike) -> Optional[bool]: ...
447 def get_boolean(
448 self, section: SectionLike, name: NameLike, default: Optional[bool] = None
449 ) -> Optional[bool]:
450 """Retrieve a configuration setting as boolean.
452 Args:
453 section: Tuple with section name and optional subsection name
454 name: Name of the setting, including section and possible
455 subsection.
456 default: Default value if setting is not found
458 Returns:
459 Contents of the setting
460 """
461 try:
462 value = self.get(section, name)
463 except KeyError:
464 return default
465 if value.lower() == b"true":
466 return True
467 elif value.lower() == b"false":
468 return False
469 raise ValueError(f"not a valid boolean string: {value!r}")
471 def set(
472 self, section: SectionLike, name: NameLike, value: Union[ValueLike, bool]
473 ) -> None:
474 """Set a configuration value.
476 Args:
477 section: Tuple with section name and optional subsection namee
478 name: Name of the configuration value, including section
479 and optional subsection
480 value: value of the setting
481 """
482 raise NotImplementedError(self.set)
484 def items(self, section: SectionLike) -> Iterator[tuple[Name, Value]]:
485 """Iterate over the configuration pairs for a specific section.
487 Args:
488 section: Tuple with section name and optional subsection namee
489 Returns:
490 Iterator over (name, value) pairs
491 """
492 raise NotImplementedError(self.items)
494 def sections(self) -> Iterator[Section]:
495 """Iterate over the sections.
497 Returns: Iterator over section tuples
498 """
499 raise NotImplementedError(self.sections)
501 def has_section(self, name: Section) -> bool:
502 """Check if a specified section exists.
504 Args:
505 name: Name of section to check for
506 Returns:
507 boolean indicating whether the section exists
508 """
509 return name in self.sections()
512class ConfigDict(Config):
513 """Git configuration stored in a dictionary."""
515 def __init__(
516 self,
517 values: Union[
518 MutableMapping[Section, CaseInsensitiveOrderedMultiDict[Name, Value]], None
519 ] = None,
520 encoding: Union[str, None] = None,
521 ) -> None:
522 """Create a new ConfigDict."""
523 if encoding is None:
524 encoding = sys.getdefaultencoding()
525 self.encoding = encoding
526 self._values: CaseInsensitiveOrderedMultiDict[
527 Section, CaseInsensitiveOrderedMultiDict[Name, Value]
528 ] = CaseInsensitiveOrderedMultiDict.make(
529 values, default_factory=CaseInsensitiveOrderedMultiDict
530 )
532 def __repr__(self) -> str:
533 """Return string representation of ConfigDict."""
534 return f"{self.__class__.__name__}({self._values!r})"
536 def __eq__(self, other: object) -> bool:
537 """Check equality with another ConfigDict."""
538 return isinstance(other, self.__class__) and other._values == self._values
540 def __getitem__(self, key: Section) -> CaseInsensitiveOrderedMultiDict[Name, Value]:
541 """Get configuration values for a section.
543 Raises:
544 KeyError: If section not found
545 """
546 return self._values.__getitem__(key)
548 def __setitem__(
549 self, key: Section, value: CaseInsensitiveOrderedMultiDict[Name, Value]
550 ) -> None:
551 """Set configuration values for a section."""
552 return self._values.__setitem__(key, value)
554 def __delitem__(self, key: Section) -> None:
555 """Delete a configuration section.
557 Raises:
558 KeyError: If section not found
559 """
560 return self._values.__delitem__(key)
562 def __iter__(self) -> Iterator[Section]:
563 """Iterate over configuration sections."""
564 return self._values.__iter__()
566 def __len__(self) -> int:
567 """Return the number of sections."""
568 return self._values.__len__()
570 def keys(self) -> KeysView[Section]:
571 """Return a view of section names."""
572 return self._values.keys()
574 @classmethod
575 def _parse_setting(cls, name: str) -> tuple[str, Optional[str], str]:
576 parts = name.split(".")
577 if len(parts) == 3:
578 return (parts[0], parts[1], parts[2])
579 else:
580 return (parts[0], None, parts[1])
582 def _check_section_and_name(
583 self, section: SectionLike, name: NameLike
584 ) -> tuple[Section, Name]:
585 if not isinstance(section, tuple):
586 section = (section,)
588 checked_section = tuple(
589 [
590 subsection.encode(self.encoding)
591 if not isinstance(subsection, bytes)
592 else subsection
593 for subsection in section
594 ]
595 )
597 if not isinstance(name, bytes):
598 name = name.encode(self.encoding)
600 return checked_section, name
602 def get_multivar(self, section: SectionLike, name: NameLike) -> Iterator[Value]:
603 """Get multiple values for a configuration setting.
605 Args:
606 section: Section name
607 name: Setting name
609 Returns:
610 Iterator of configuration values
611 """
612 section, name = self._check_section_and_name(section, name)
614 if len(section) > 1:
615 try:
616 return self._values[section].get_all(name)
617 except KeyError:
618 pass
620 return self._values[(section[0],)].get_all(name)
622 def get(
623 self,
624 section: SectionLike,
625 name: NameLike,
626 ) -> Value:
627 """Get a configuration value.
629 Args:
630 section: Section name
631 name: Setting name
633 Returns:
634 Configuration value
636 Raises:
637 KeyError: if the value is not set
638 """
639 section, name = self._check_section_and_name(section, name)
641 if len(section) > 1:
642 try:
643 return self._values[section][name]
644 except KeyError:
645 pass
647 return self._values[(section[0],)][name]
649 def set(
650 self,
651 section: SectionLike,
652 name: NameLike,
653 value: Union[ValueLike, bool],
654 ) -> None:
655 """Set a configuration value.
657 Args:
658 section: Section name
659 name: Setting name
660 value: Configuration value
661 """
662 section, name = self._check_section_and_name(section, name)
664 if isinstance(value, bool):
665 value = b"true" if value else b"false"
667 if not isinstance(value, bytes):
668 value = value.encode(self.encoding)
670 section_dict = self._values.setdefault(section)
671 if hasattr(section_dict, "set"):
672 section_dict.set(name, value)
673 else:
674 section_dict[name] = value
676 def add(
677 self,
678 section: SectionLike,
679 name: NameLike,
680 value: Union[ValueLike, bool],
681 ) -> None:
682 """Add a value to a configuration setting, creating a multivar if needed."""
683 section, name = self._check_section_and_name(section, name)
685 if isinstance(value, bool):
686 value = b"true" if value else b"false"
688 if not isinstance(value, bytes):
689 value = value.encode(self.encoding)
691 self._values.setdefault(section)[name] = value
693 def items(self, section: SectionLike) -> Iterator[tuple[Name, Value]]:
694 """Get items in a section."""
695 section_bytes, _ = self._check_section_and_name(section, b"")
696 section_dict = self._values.get(section_bytes)
697 if section_dict is not None:
698 return iter(section_dict.items())
699 return iter([])
701 def sections(self) -> Iterator[Section]:
702 """Get all sections."""
703 return iter(self._values.keys())
706def _format_string(value: bytes) -> bytes:
707 if (
708 value.startswith((b" ", b"\t"))
709 or value.endswith((b" ", b"\t"))
710 or b"#" in value
711 ):
712 return b'"' + _escape_value(value) + b'"'
713 else:
714 return _escape_value(value)
717_ESCAPE_TABLE = {
718 ord(b"\\"): ord(b"\\"),
719 ord(b'"'): ord(b'"'),
720 ord(b"n"): ord(b"\n"),
721 ord(b"t"): ord(b"\t"),
722 ord(b"b"): ord(b"\b"),
723}
724_COMMENT_CHARS = [ord(b"#"), ord(b";")]
725_WHITESPACE_CHARS = [ord(b"\t"), ord(b" ")]
728def _parse_string(value: bytes) -> bytes:
729 value = bytearray(value.strip())
730 ret = bytearray()
731 whitespace = bytearray()
732 in_quotes = False
733 i = 0
734 while i < len(value):
735 c = value[i]
736 if c == ord(b"\\"):
737 i += 1
738 if i >= len(value):
739 # Backslash at end of string - treat as literal backslash
740 if whitespace:
741 ret.extend(whitespace)
742 whitespace = bytearray()
743 ret.append(ord(b"\\"))
744 else:
745 try:
746 v = _ESCAPE_TABLE[value[i]]
747 if whitespace:
748 ret.extend(whitespace)
749 whitespace = bytearray()
750 ret.append(v)
751 except KeyError:
752 # Unknown escape sequence - treat backslash as literal and process next char normally
753 if whitespace:
754 ret.extend(whitespace)
755 whitespace = bytearray()
756 ret.append(ord(b"\\"))
757 i -= 1 # Reprocess the character after the backslash
758 elif c == ord(b'"'):
759 in_quotes = not in_quotes
760 elif c in _COMMENT_CHARS and not in_quotes:
761 # the rest of the line is a comment
762 break
763 elif c in _WHITESPACE_CHARS:
764 whitespace.append(c)
765 else:
766 if whitespace:
767 ret.extend(whitespace)
768 whitespace = bytearray()
769 ret.append(c)
770 i += 1
772 if in_quotes:
773 raise ValueError("missing end quote")
775 return bytes(ret)
778def _escape_value(value: bytes) -> bytes:
779 """Escape a value."""
780 value = value.replace(b"\\", b"\\\\")
781 value = value.replace(b"\r", b"\\r")
782 value = value.replace(b"\n", b"\\n")
783 value = value.replace(b"\t", b"\\t")
784 value = value.replace(b'"', b'\\"')
785 return value
788def _check_variable_name(name: bytes) -> bool:
789 for i in range(len(name)):
790 c = name[i : i + 1]
791 if not c.isalnum() and c != b"-":
792 return False
793 return True
796def _check_section_name(name: bytes) -> bool:
797 for i in range(len(name)):
798 c = name[i : i + 1]
799 if not c.isalnum() and c not in (b"-", b"."):
800 return False
801 return True
804def _strip_comments(line: bytes) -> bytes:
805 comment_bytes = {ord(b"#"), ord(b";")}
806 quote = ord(b'"')
807 string_open = False
808 # Normalize line to bytearray for simple 2/3 compatibility
809 for i, character in enumerate(bytearray(line)):
810 # Comment characters outside balanced quotes denote comment start
811 if character == quote:
812 string_open = not string_open
813 elif not string_open and character in comment_bytes:
814 return line[:i]
815 return line
818def _is_line_continuation(value: bytes) -> bool:
819 """Check if a value ends with a line continuation backslash.
821 A line continuation occurs when a line ends with a backslash that is:
822 1. Not escaped (not preceded by another backslash)
823 2. Not within quotes
825 Args:
826 value: The value to check
828 Returns:
829 True if the value ends with a line continuation backslash
830 """
831 if not value.endswith((b"\\\n", b"\\\r\n")):
832 return False
834 # Remove only the newline characters, keep the content including the backslash
835 if value.endswith(b"\\\r\n"):
836 content = value[:-2] # Remove \r\n, keep the \
837 else:
838 content = value[:-1] # Remove \n, keep the \
840 if not content.endswith(b"\\"):
841 return False
843 # Count consecutive backslashes at the end
844 backslash_count = 0
845 for i in range(len(content) - 1, -1, -1):
846 if content[i : i + 1] == b"\\":
847 backslash_count += 1
848 else:
849 break
851 # If we have an odd number of backslashes, the last one is a line continuation
852 # If we have an even number, they are all escaped and there's no continuation
853 return backslash_count % 2 == 1
856def _parse_section_header_line(line: bytes) -> tuple[Section, bytes]:
857 # Parse section header ("[bla]")
858 line = _strip_comments(line).rstrip()
859 in_quotes = False
860 escaped = False
861 for i, c in enumerate(line):
862 if escaped:
863 escaped = False
864 continue
865 if c == ord(b'"'):
866 in_quotes = not in_quotes
867 if c == ord(b"\\"):
868 escaped = True
869 if c == ord(b"]") and not in_quotes:
870 last = i
871 break
872 else:
873 raise ValueError("expected trailing ]")
874 pts = line[1:last].split(b" ", 1)
875 line = line[last + 1 :]
876 section: Section
877 if len(pts) == 2:
878 # Handle subsections - Git allows more complex syntax for certain sections like includeIf
879 if pts[1][:1] == b'"' and pts[1][-1:] == b'"':
880 # Standard quoted subsection
881 pts[1] = pts[1][1:-1]
882 elif pts[0] == b"includeIf":
883 # Special handling for includeIf sections which can have complex conditions
884 # Git allows these without strict quote validation
885 pts[1] = pts[1].strip()
886 if pts[1][:1] == b'"' and pts[1][-1:] == b'"':
887 pts[1] = pts[1][1:-1]
888 else:
889 # Other sections must have quoted subsections
890 raise ValueError(f"Invalid subsection {pts[1]!r}")
891 if not _check_section_name(pts[0]):
892 raise ValueError(f"invalid section name {pts[0]!r}")
893 section = (pts[0], pts[1])
894 else:
895 if not _check_section_name(pts[0]):
896 raise ValueError(f"invalid section name {pts[0]!r}")
897 pts = pts[0].split(b".", 1)
898 if len(pts) == 2:
899 section = (pts[0], pts[1])
900 else:
901 section = (pts[0],)
902 return section, line
905class ConfigFile(ConfigDict):
906 """A Git configuration file, like .git/config or ~/.gitconfig."""
908 def __init__(
909 self,
910 values: Union[
911 MutableMapping[Section, CaseInsensitiveOrderedMultiDict[Name, Value]], None
912 ] = None,
913 encoding: Union[str, None] = None,
914 ) -> None:
915 """Initialize a ConfigFile.
917 Args:
918 values: Optional mapping of configuration values
919 encoding: Optional encoding for the file (defaults to system encoding)
920 """
921 super().__init__(values=values, encoding=encoding)
922 self.path: Optional[str] = None
923 self._included_paths: set[str] = set() # Track included files to prevent cycles
925 @classmethod
926 def from_file(
927 cls,
928 f: IO[bytes],
929 *,
930 config_dir: Optional[str] = None,
931 included_paths: Optional[set[str]] = None,
932 include_depth: int = 0,
933 max_include_depth: int = DEFAULT_MAX_INCLUDE_DEPTH,
934 file_opener: Optional[FileOpener] = None,
935 condition_matchers: Optional[dict[str, ConditionMatcher]] = None,
936 ) -> "ConfigFile":
937 """Read configuration from a file-like object.
939 Args:
940 f: File-like object to read from
941 config_dir: Directory containing the config file (for relative includes)
942 included_paths: Set of already included paths (to prevent cycles)
943 include_depth: Current include depth (to prevent infinite recursion)
944 max_include_depth: Maximum allowed include depth
945 file_opener: Optional callback to open included files
946 condition_matchers: Optional dict of condition matchers for includeIf
947 """
948 if include_depth > max_include_depth:
949 # Prevent excessive recursion
950 raise ValueError(f"Maximum include depth ({max_include_depth}) exceeded")
952 ret = cls()
953 if included_paths is not None:
954 ret._included_paths = included_paths.copy()
956 section: Optional[Section] = None
957 setting = None
958 continuation = None
959 for lineno, line in enumerate(f.readlines()):
960 if lineno == 0 and line.startswith(b"\xef\xbb\xbf"):
961 line = line[3:]
962 line = line.lstrip()
963 if setting is None:
964 if len(line) > 0 and line[:1] == b"[":
965 section, line = _parse_section_header_line(line)
966 ret._values.setdefault(section)
967 if _strip_comments(line).strip() == b"":
968 continue
969 if section is None:
970 raise ValueError(f"setting {line!r} without section")
971 try:
972 setting, value = line.split(b"=", 1)
973 except ValueError:
974 setting = line
975 value = b"true"
976 setting = setting.strip()
977 if not _check_variable_name(setting):
978 raise ValueError(f"invalid variable name {setting!r}")
979 if _is_line_continuation(value):
980 if value.endswith(b"\\\r\n"):
981 continuation = value[:-3]
982 else:
983 continuation = value[:-2]
984 else:
985 continuation = None
986 value = _parse_string(value)
987 ret._values[section][setting] = value
989 # Process include/includeIf directives
990 ret._handle_include_directive(
991 section,
992 setting,
993 value,
994 config_dir=config_dir,
995 include_depth=include_depth,
996 max_include_depth=max_include_depth,
997 file_opener=file_opener,
998 condition_matchers=condition_matchers,
999 )
1001 setting = None
1002 else: # continuation line
1003 assert continuation is not None
1004 if _is_line_continuation(line):
1005 if line.endswith(b"\\\r\n"):
1006 continuation += line[:-3]
1007 else:
1008 continuation += line[:-2]
1009 else:
1010 continuation += line
1011 value = _parse_string(continuation)
1012 assert section is not None # Already checked above
1013 ret._values[section][setting] = value
1015 # Process include/includeIf directives
1016 ret._handle_include_directive(
1017 section,
1018 setting,
1019 value,
1020 config_dir=config_dir,
1021 include_depth=include_depth,
1022 max_include_depth=max_include_depth,
1023 file_opener=file_opener,
1024 condition_matchers=condition_matchers,
1025 )
1027 continuation = None
1028 setting = None
1029 return ret
1031 def _handle_include_directive(
1032 self,
1033 section: Optional[Section],
1034 setting: bytes,
1035 value: bytes,
1036 *,
1037 config_dir: Optional[str],
1038 include_depth: int,
1039 max_include_depth: int,
1040 file_opener: Optional[FileOpener],
1041 condition_matchers: Optional[dict[str, ConditionMatcher]],
1042 ) -> None:
1043 """Handle include/includeIf directives during config parsing."""
1044 if (
1045 section is not None
1046 and setting == b"path"
1047 and (
1048 section[0].lower() == b"include"
1049 or (len(section) > 1 and section[0].lower() == b"includeif")
1050 )
1051 ):
1052 self._process_include(
1053 section,
1054 value,
1055 config_dir=config_dir,
1056 include_depth=include_depth,
1057 max_include_depth=max_include_depth,
1058 file_opener=file_opener,
1059 condition_matchers=condition_matchers,
1060 )
1062 def _process_include(
1063 self,
1064 section: Section,
1065 path_value: bytes,
1066 *,
1067 config_dir: Optional[str],
1068 include_depth: int,
1069 max_include_depth: int,
1070 file_opener: Optional[FileOpener],
1071 condition_matchers: Optional[dict[str, ConditionMatcher]],
1072 ) -> None:
1073 """Process an include or includeIf directive."""
1074 path_str = path_value.decode(self.encoding, errors="replace")
1076 # Handle includeIf conditions
1077 if len(section) > 1 and section[0].lower() == b"includeif":
1078 condition = section[1].decode(self.encoding, errors="replace")
1079 if not self._evaluate_includeif_condition(
1080 condition, config_dir, condition_matchers
1081 ):
1082 return
1084 # Resolve the include path
1085 include_path = self._resolve_include_path(path_str, config_dir)
1086 if not include_path:
1087 return
1089 # Check for circular includes
1090 try:
1091 abs_path = str(Path(include_path).resolve())
1092 except (OSError, ValueError) as e:
1093 # Invalid path - log and skip
1094 logger.debug("Invalid include path %r: %s", include_path, e)
1095 return
1096 if abs_path in self._included_paths:
1097 return
1099 # Load and merge the included file
1100 try:
1101 # Use provided file opener or default to GitFile
1102 opener: FileOpener
1103 if file_opener is None:
1105 def opener(path: Union[str, os.PathLike]) -> IO[bytes]:
1106 return GitFile(path, "rb")
1107 else:
1108 opener = file_opener
1110 f = opener(include_path)
1111 except (OSError, ValueError) as e:
1112 # Git silently ignores missing or unreadable include files
1113 # Log for debugging purposes
1114 logger.debug("Invalid include path %r: %s", include_path, e)
1115 else:
1116 with f as included_file:
1117 # Track this path to prevent cycles
1118 self._included_paths.add(abs_path)
1120 # Parse the included file
1121 included_config = ConfigFile.from_file(
1122 included_file,
1123 config_dir=os.path.dirname(include_path),
1124 included_paths=self._included_paths,
1125 include_depth=include_depth + 1,
1126 max_include_depth=max_include_depth,
1127 file_opener=file_opener,
1128 condition_matchers=condition_matchers,
1129 )
1131 # Merge the included configuration
1132 self._merge_config(included_config)
1134 def _merge_config(self, other: "ConfigFile") -> None:
1135 """Merge another config file into this one."""
1136 for section, values in other._values.items():
1137 if section not in self._values:
1138 self._values[section] = CaseInsensitiveOrderedMultiDict()
1139 for key, value in values.items():
1140 self._values[section][key] = value
1142 def _resolve_include_path(
1143 self, path: str, config_dir: Optional[str]
1144 ) -> Optional[str]:
1145 """Resolve an include path to an absolute path."""
1146 # Expand ~ to home directory
1147 path = os.path.expanduser(path)
1149 # If path is relative and we have a config directory, make it relative to that
1150 if not os.path.isabs(path) and config_dir:
1151 path = os.path.join(config_dir, path)
1153 return path
1155 def _evaluate_includeif_condition(
1156 self,
1157 condition: str,
1158 config_dir: Optional[str] = None,
1159 condition_matchers: Optional[dict[str, ConditionMatcher]] = None,
1160 ) -> bool:
1161 """Evaluate an includeIf condition."""
1162 # Try custom matchers first if provided
1163 if condition_matchers:
1164 for prefix, matcher in condition_matchers.items():
1165 if condition.startswith(prefix):
1166 return matcher(condition[len(prefix) :])
1168 # Fall back to built-in matchers
1169 if condition.startswith("hasconfig:"):
1170 return self._evaluate_hasconfig_condition(condition[10:])
1171 else:
1172 # Unknown condition type - log and ignore (Git behavior)
1173 logger.debug("Unknown includeIf condition: %r", condition)
1174 return False
1176 def _evaluate_hasconfig_condition(self, condition: str) -> bool:
1177 """Evaluate a hasconfig condition.
1179 Format: hasconfig:config.key:pattern
1180 Example: hasconfig:remote.*.url:ssh://org-*@github.com/**
1181 """
1182 # Split on the first colon to separate config key from pattern
1183 parts = condition.split(":", 1)
1184 if len(parts) != 2:
1185 logger.debug("Invalid hasconfig condition format: %r", condition)
1186 return False
1188 config_key, pattern = parts
1190 # Parse the config key to get section and name
1191 key_parts = config_key.split(".", 2)
1192 if len(key_parts) < 2:
1193 logger.debug("Invalid hasconfig config key: %r", config_key)
1194 return False
1196 # Handle wildcards in section names (e.g., remote.*)
1197 if len(key_parts) == 3 and key_parts[1] == "*":
1198 # Match any subsection
1199 section_prefix = key_parts[0].encode(self.encoding)
1200 name = key_parts[2].encode(self.encoding)
1202 # Check all sections that match the pattern
1203 for section in self.sections():
1204 if len(section) == 2 and section[0] == section_prefix:
1205 try:
1206 values = list(self.get_multivar(section, name))
1207 for value in values:
1208 if self._match_hasconfig_pattern(value, pattern):
1209 return True
1210 except KeyError:
1211 continue
1212 else:
1213 # Direct section lookup
1214 if len(key_parts) == 2:
1215 section = (key_parts[0].encode(self.encoding),)
1216 name = key_parts[1].encode(self.encoding)
1217 else:
1218 section = (
1219 key_parts[0].encode(self.encoding),
1220 key_parts[1].encode(self.encoding),
1221 )
1222 name = key_parts[2].encode(self.encoding)
1224 try:
1225 values = list(self.get_multivar(section, name))
1226 for value in values:
1227 if self._match_hasconfig_pattern(value, pattern):
1228 return True
1229 except KeyError:
1230 pass
1232 return False
1234 def _match_hasconfig_pattern(self, value: bytes, pattern: str) -> bool:
1235 """Match a config value against a hasconfig pattern.
1237 Supports simple glob patterns like ``*`` and ``**``.
1238 """
1239 value_str = value.decode(self.encoding, errors="replace")
1240 return match_glob_pattern(value_str, pattern)
1242 @classmethod
1243 def from_path(
1244 cls,
1245 path: Union[str, os.PathLike],
1246 *,
1247 max_include_depth: int = DEFAULT_MAX_INCLUDE_DEPTH,
1248 file_opener: Optional[FileOpener] = None,
1249 condition_matchers: Optional[dict[str, ConditionMatcher]] = None,
1250 ) -> "ConfigFile":
1251 """Read configuration from a file on disk.
1253 Args:
1254 path: Path to the configuration file
1255 max_include_depth: Maximum allowed include depth
1256 file_opener: Optional callback to open included files
1257 condition_matchers: Optional dict of condition matchers for includeIf
1258 """
1259 abs_path = os.fspath(path)
1260 config_dir = os.path.dirname(abs_path)
1262 # Use provided file opener or default to GitFile
1263 opener: FileOpener
1264 if file_opener is None:
1266 def opener(p: Union[str, os.PathLike]) -> IO[bytes]:
1267 return GitFile(p, "rb")
1268 else:
1269 opener = file_opener
1271 with opener(abs_path) as f:
1272 ret = cls.from_file(
1273 f,
1274 config_dir=config_dir,
1275 max_include_depth=max_include_depth,
1276 file_opener=file_opener,
1277 condition_matchers=condition_matchers,
1278 )
1279 ret.path = abs_path
1280 return ret
1282 def write_to_path(self, path: Optional[Union[str, os.PathLike]] = None) -> None:
1283 """Write configuration to a file on disk."""
1284 if path is None:
1285 if self.path is None:
1286 raise ValueError("No path specified and no default path available")
1287 path_to_use: Union[str, os.PathLike] = self.path
1288 else:
1289 path_to_use = path
1290 with GitFile(path_to_use, "wb") as f:
1291 self.write_to_file(f)
1293 def write_to_file(self, f: Union[IO[bytes], _GitFile]) -> None:
1294 """Write configuration to a file-like object."""
1295 for section, values in self._values.items():
1296 try:
1297 section_name, subsection_name = section
1298 except ValueError:
1299 (section_name,) = section
1300 subsection_name = None
1301 if subsection_name is None:
1302 f.write(b"[" + section_name + b"]\n")
1303 else:
1304 f.write(b"[" + section_name + b' "' + subsection_name + b'"]\n')
1305 for key, value in values.items():
1306 value = _format_string(value)
1307 f.write(b"\t" + key + b" = " + value + b"\n")
1310def get_xdg_config_home_path(*path_segments: str) -> str:
1311 """Get a path in the XDG config home directory.
1313 Args:
1314 *path_segments: Path segments to join to the XDG config home
1316 Returns:
1317 Full path in XDG config home directory
1318 """
1319 xdg_config_home = os.environ.get(
1320 "XDG_CONFIG_HOME",
1321 os.path.expanduser("~/.config/"),
1322 )
1323 return os.path.join(xdg_config_home, *path_segments)
1326def _find_git_in_win_path() -> Iterator[str]:
1327 for exe in ("git.exe", "git.cmd"):
1328 for path in os.environ.get("PATH", "").split(";"):
1329 if os.path.exists(os.path.join(path, exe)):
1330 # in windows native shells (powershell/cmd) exe path is
1331 # .../Git/bin/git.exe or .../Git/cmd/git.exe
1332 #
1333 # in git-bash exe path is .../Git/mingw64/bin/git.exe
1334 git_dir, _bin_dir = os.path.split(path)
1335 yield git_dir
1336 parent_dir, basename = os.path.split(git_dir)
1337 if basename == "mingw32" or basename == "mingw64":
1338 yield parent_dir
1339 break
1342def _find_git_in_win_reg() -> Iterator[str]:
1343 import platform
1344 import winreg
1346 if platform.machine() == "AMD64":
1347 subkey = (
1348 "SOFTWARE\\Wow6432Node\\Microsoft\\Windows\\"
1349 "CurrentVersion\\Uninstall\\Git_is1"
1350 )
1351 else:
1352 subkey = "SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\Uninstall\\Git_is1"
1354 for key in (winreg.HKEY_CURRENT_USER, winreg.HKEY_LOCAL_MACHINE): # type: ignore
1355 with suppress(OSError):
1356 with winreg.OpenKey(key, subkey) as k: # type: ignore
1357 val, typ = winreg.QueryValueEx(k, "InstallLocation") # type: ignore
1358 if typ == winreg.REG_SZ: # type: ignore
1359 yield val
1362# There is no set standard for system config dirs on windows. We try the
1363# following:
1364# - %PROGRAMFILES%/Git/etc/gitconfig - Git for Windows (msysgit) config dir
1365# Used if CGit installation (Git/bin/git.exe) is found in PATH in the
1366# system registry
1367def get_win_system_paths() -> Iterator[str]:
1368 """Get current Windows system Git config paths.
1370 Only returns the current Git for Windows config location, not legacy paths.
1371 """
1372 # Try to find Git installation from PATH first
1373 for git_dir in _find_git_in_win_path():
1374 yield os.path.join(git_dir, "etc", "gitconfig")
1375 return # Only use the first found path
1377 # Fall back to registry if not found in PATH
1378 for git_dir in _find_git_in_win_reg():
1379 yield os.path.join(git_dir, "etc", "gitconfig")
1380 return # Only use the first found path
1383def get_win_legacy_system_paths() -> Iterator[str]:
1384 """Get legacy Windows system Git config paths.
1386 Returns all possible config paths including deprecated locations.
1387 This function can be used for diagnostics or migration purposes.
1388 """
1389 # Include deprecated PROGRAMDATA location
1390 if "PROGRAMDATA" in os.environ:
1391 yield os.path.join(os.environ["PROGRAMDATA"], "Git", "config")
1393 # Include all Git installations found
1394 for git_dir in _find_git_in_win_path():
1395 yield os.path.join(git_dir, "etc", "gitconfig")
1396 for git_dir in _find_git_in_win_reg():
1397 yield os.path.join(git_dir, "etc", "gitconfig")
1400class StackedConfig(Config):
1401 """Configuration which reads from multiple config files.."""
1403 def __init__(
1404 self, backends: list[ConfigFile], writable: Optional[ConfigFile] = None
1405 ) -> None:
1406 """Initialize a StackedConfig.
1408 Args:
1409 backends: List of config files to read from (in order of precedence)
1410 writable: Optional config file to write changes to
1411 """
1412 self.backends = backends
1413 self.writable = writable
1415 def __repr__(self) -> str:
1416 """Return string representation of StackedConfig."""
1417 return f"<{self.__class__.__name__} for {self.backends!r}>"
1419 @classmethod
1420 def default(cls) -> "StackedConfig":
1421 """Create a StackedConfig with default system/user config files.
1423 Returns:
1424 StackedConfig with default configuration files loaded
1425 """
1426 return cls(cls.default_backends())
1428 @classmethod
1429 def default_backends(cls) -> list[ConfigFile]:
1430 """Retrieve the default configuration.
1432 See git-config(1) for details on the files searched.
1433 """
1434 paths = []
1436 # Handle GIT_CONFIG_GLOBAL - overrides user config paths
1437 try:
1438 paths.append(os.environ["GIT_CONFIG_GLOBAL"])
1439 except KeyError:
1440 paths.append(os.path.expanduser("~/.gitconfig"))
1441 paths.append(get_xdg_config_home_path("git", "config"))
1443 # Handle GIT_CONFIG_SYSTEM and GIT_CONFIG_NOSYSTEM
1444 try:
1445 paths.append(os.environ["GIT_CONFIG_SYSTEM"])
1446 except KeyError:
1447 if "GIT_CONFIG_NOSYSTEM" not in os.environ:
1448 paths.append("/etc/gitconfig")
1449 if sys.platform == "win32":
1450 paths.extend(get_win_system_paths())
1452 logger.debug("Loading gitconfig from paths: %s", paths)
1454 backends = []
1455 for path in paths:
1456 try:
1457 cf = ConfigFile.from_path(path)
1458 logger.debug("Successfully loaded gitconfig from: %s", path)
1459 except FileNotFoundError:
1460 logger.debug("Gitconfig file not found: %s", path)
1461 continue
1462 backends.append(cf)
1463 return backends
1465 def get(self, section: SectionLike, name: NameLike) -> Value:
1466 """Get value from configuration."""
1467 if not isinstance(section, tuple):
1468 section = (section,)
1469 for backend in self.backends:
1470 try:
1471 return backend.get(section, name)
1472 except KeyError:
1473 pass
1474 raise KeyError(name)
1476 def get_multivar(self, section: SectionLike, name: NameLike) -> Iterator[Value]:
1477 """Get multiple values from configuration."""
1478 if not isinstance(section, tuple):
1479 section = (section,)
1480 for backend in self.backends:
1481 try:
1482 yield from backend.get_multivar(section, name)
1483 except KeyError:
1484 pass
1486 def set(
1487 self, section: SectionLike, name: NameLike, value: Union[ValueLike, bool]
1488 ) -> None:
1489 """Set value in configuration."""
1490 if self.writable is None:
1491 raise NotImplementedError(self.set)
1492 return self.writable.set(section, name, value)
1494 def sections(self) -> Iterator[Section]:
1495 """Get all sections."""
1496 seen = set()
1497 for backend in self.backends:
1498 for section in backend.sections():
1499 if section not in seen:
1500 seen.add(section)
1501 yield section
1504def read_submodules(
1505 path: Union[str, os.PathLike],
1506) -> Iterator[tuple[bytes, bytes, bytes]]:
1507 """Read a .gitmodules file."""
1508 cfg = ConfigFile.from_path(path)
1509 return parse_submodules(cfg)
1512def parse_submodules(config: ConfigFile) -> Iterator[tuple[bytes, bytes, bytes]]:
1513 """Parse a gitmodules GitConfig file, returning submodules.
1515 Args:
1516 config: A `ConfigFile`
1517 Returns:
1518 list of tuples (submodule path, url, name),
1519 where name is quoted part of the section's name.
1520 """
1521 for section in config.sections():
1522 section_kind, section_name = section
1523 if section_kind == b"submodule":
1524 try:
1525 sm_path = config.get(section, b"path")
1526 sm_url = config.get(section, b"url")
1527 yield (sm_path, sm_url, section_name)
1528 except KeyError:
1529 # If either path or url is missing, just ignore this
1530 # submodule entry and move on to the next one. This is
1531 # how git itself handles malformed .gitmodule entries.
1532 pass
1535def iter_instead_of(config: Config, push: bool = False) -> Iterable[tuple[str, str]]:
1536 """Iterate over insteadOf / pushInsteadOf values."""
1537 for section in config.sections():
1538 if section[0] != b"url":
1539 continue
1540 replacement = section[1]
1541 try:
1542 needles = list(config.get_multivar(section, "insteadOf"))
1543 except KeyError:
1544 needles = []
1545 if push:
1546 try:
1547 needles += list(config.get_multivar(section, "pushInsteadOf"))
1548 except KeyError:
1549 pass
1550 for needle in needles:
1551 assert isinstance(needle, bytes)
1552 yield needle.decode("utf-8"), replacement.decode("utf-8")
1555def apply_instead_of(config: Config, orig_url: str, push: bool = False) -> str:
1556 """Apply insteadOf / pushInsteadOf to a URL."""
1557 longest_needle = ""
1558 updated_url = orig_url
1559 for needle, replacement in iter_instead_of(config, push):
1560 if not orig_url.startswith(needle):
1561 continue
1562 if len(longest_needle) < len(needle):
1563 longest_needle = needle
1564 updated_url = replacement + orig_url[len(needle) :]
1565 return updated_url