Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/dulwich/config.py: 63%
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 public 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 Any,
44 BinaryIO,
45 Callable,
46 Generic,
47 Optional,
48 TypeVar,
49 Union,
50 overload,
51)
53from .file import GitFile
55ConfigKey = Union[str, bytes, tuple[Union[str, bytes], ...]]
56ConfigValue = Union[str, bytes, bool, int]
58logger = logging.getLogger(__name__)
60# Type for file opener callback
61FileOpener = Callable[[Union[str, os.PathLike]], BinaryIO]
63# Type for includeIf condition matcher
64# Takes the condition value (e.g., "main" for onbranch:main) and returns bool
65ConditionMatcher = Callable[[str], bool]
67# Security limits for include files
68MAX_INCLUDE_FILE_SIZE = 1024 * 1024 # 1MB max for included config files
69DEFAULT_MAX_INCLUDE_DEPTH = 10 # Maximum recursion depth for includes
72def _match_gitdir_pattern(
73 path: bytes, pattern: bytes, ignorecase: bool = False
74) -> bool:
75 """Simple gitdir pattern matching for includeIf conditions.
77 This handles the basic gitdir patterns used in includeIf directives.
78 """
79 # Convert to strings for easier manipulation
80 path_str = path.decode("utf-8", errors="replace")
81 pattern_str = pattern.decode("utf-8", errors="replace")
83 # Normalize paths to use forward slashes for consistent matching
84 path_str = path_str.replace("\\", "/")
85 pattern_str = pattern_str.replace("\\", "/")
87 if ignorecase:
88 path_str = path_str.lower()
89 pattern_str = pattern_str.lower()
91 # Handle the common cases for gitdir patterns
92 if pattern_str.startswith("**/") and pattern_str.endswith("/**"):
93 # Pattern like **/dirname/** should match any path containing dirname
94 dirname = pattern_str[3:-3] # Remove **/ and /**
95 # Check if path contains the directory name as a path component
96 return ("/" + dirname + "/") in path_str or path_str.endswith("/" + dirname)
97 elif pattern_str.startswith("**/"):
98 # Pattern like **/filename
99 suffix = pattern_str[3:] # Remove **/
100 return suffix in path_str or path_str.endswith("/" + suffix)
101 elif pattern_str.endswith("/**"):
102 # Pattern like /path/to/dir/** should match /path/to/dir and any subdirectory
103 base_pattern = pattern_str[:-3] # Remove /**
104 return path_str == base_pattern or path_str.startswith(base_pattern + "/")
105 elif "**" in pattern_str:
106 # Handle patterns with ** in the middle
107 parts = pattern_str.split("**")
108 if len(parts) == 2:
109 prefix, suffix = parts
110 # Path must start with prefix and end with suffix (if any)
111 if prefix and not path_str.startswith(prefix):
112 return False
113 if suffix and not path_str.endswith(suffix):
114 return False
115 return True
117 # Direct match or simple glob pattern
118 if "*" in pattern_str or "?" in pattern_str or "[" in pattern_str:
119 import fnmatch
121 return fnmatch.fnmatch(path_str, pattern_str)
122 else:
123 return path_str == pattern_str
126def match_glob_pattern(value: str, pattern: str) -> bool:
127 r"""Match a value against a glob pattern.
129 Supports simple glob patterns like ``*`` and ``**``.
131 Raises:
132 ValueError: If the pattern is invalid
133 """
134 # Convert glob pattern to regex
135 pattern_escaped = re.escape(pattern)
136 # Replace escaped \*\* with .* (match anything)
137 pattern_escaped = pattern_escaped.replace(r"\*\*", ".*")
138 # Replace escaped \* with [^/]* (match anything except /)
139 pattern_escaped = pattern_escaped.replace(r"\*", "[^/]*")
140 # Anchor the pattern
141 pattern_regex = f"^{pattern_escaped}$"
143 try:
144 return bool(re.match(pattern_regex, value))
145 except re.error as e:
146 raise ValueError(f"Invalid glob pattern {pattern!r}: {e}")
149def lower_key(key: ConfigKey) -> ConfigKey:
150 if isinstance(key, (bytes, str)):
151 return key.lower()
153 if isinstance(key, tuple):
154 # For config sections, only lowercase the section name (first element)
155 # but preserve the case of subsection names (remaining elements)
156 if len(key) > 0:
157 first = key[0]
158 assert isinstance(first, (bytes, str))
159 return (first.lower(), *key[1:])
160 return key
162 raise TypeError(key)
165K = TypeVar("K", bound=ConfigKey) # Key type must be ConfigKey
166V = TypeVar("V") # Value type
167_T = TypeVar("_T") # For get() default parameter
170class CaseInsensitiveOrderedMultiDict(MutableMapping[K, V], Generic[K, V]):
171 def __init__(self, default_factory: Optional[Callable[[], V]] = None) -> None:
172 self._real: list[tuple[K, V]] = []
173 self._keyed: dict[Any, V] = {}
174 self._default_factory = default_factory
176 @classmethod
177 def make(
178 cls, dict_in=None, default_factory=None
179 ) -> "CaseInsensitiveOrderedMultiDict[K, V]":
180 if isinstance(dict_in, cls):
181 return dict_in
183 out = cls(default_factory=default_factory)
185 if dict_in is None:
186 return out
188 if not isinstance(dict_in, MutableMapping):
189 raise TypeError
191 for key, value in dict_in.items():
192 out[key] = value
194 return out
196 def __len__(self) -> int:
197 return len(self._keyed)
199 def keys(self) -> KeysView[K]:
200 return self._keyed.keys() # type: ignore[return-value]
202 def items(self) -> ItemsView[K, V]:
203 # Return a view that iterates over the real list to preserve order
204 class OrderedItemsView(ItemsView[K, V]):
205 def __init__(self, mapping: CaseInsensitiveOrderedMultiDict[K, V]):
206 self._mapping = mapping
208 def __iter__(self) -> Iterator[tuple[K, V]]:
209 return iter(self._mapping._real)
211 def __len__(self) -> int:
212 return len(self._mapping._real)
214 def __contains__(self, item: object) -> bool:
215 if not isinstance(item, tuple) or len(item) != 2:
216 return False
217 key, value = item
218 return any(k == key and v == value for k, v in self._mapping._real)
220 return OrderedItemsView(self)
222 def __iter__(self) -> Iterator[K]:
223 return iter(self._keyed)
225 def values(self) -> ValuesView[V]:
226 return self._keyed.values()
228 def __setitem__(self, key, value) -> None:
229 self._real.append((key, value))
230 self._keyed[lower_key(key)] = value
232 def set(self, key, value) -> None:
233 # This method replaces all existing values for the key
234 lower = lower_key(key)
235 self._real = [(k, v) for k, v in self._real if lower_key(k) != lower]
236 self._real.append((key, value))
237 self._keyed[lower] = value
239 def __delitem__(self, key) -> None:
240 key = lower_key(key)
241 del self._keyed[key]
242 for i, (actual, unused_value) in reversed(list(enumerate(self._real))):
243 if lower_key(actual) == key:
244 del self._real[i]
246 def __getitem__(self, item: K) -> V:
247 return self._keyed[lower_key(item)]
249 def get(self, key: K, /, default: Union[V, _T, None] = None) -> Union[V, _T, None]: # type: ignore[override]
250 try:
251 return self[key]
252 except KeyError:
253 if default is not None:
254 return default
255 elif self._default_factory is not None:
256 return self._default_factory()
257 else:
258 return None
260 def get_all(self, key: K) -> Iterator[V]:
261 lowered_key = lower_key(key)
262 for actual, value in self._real:
263 if lower_key(actual) == lowered_key:
264 yield value
266 def setdefault(self, key: K, default: Optional[V] = None) -> V:
267 try:
268 return self[key]
269 except KeyError:
270 if default is not None:
271 self[key] = default
272 return default
273 elif self._default_factory is not None:
274 value = self._default_factory()
275 self[key] = value
276 return value
277 else:
278 raise
281Name = bytes
282NameLike = Union[bytes, str]
283Section = tuple[bytes, ...]
284SectionLike = Union[bytes, str, tuple[Union[bytes, str], ...]]
285Value = bytes
286ValueLike = Union[bytes, str]
289class Config:
290 """A Git configuration."""
292 def get(self, section: SectionLike, name: NameLike) -> Value:
293 """Retrieve the contents of a configuration setting.
295 Args:
296 section: Tuple with section name and optional subsection name
297 name: Variable name
298 Returns:
299 Contents of the setting
300 Raises:
301 KeyError: if the value is not set
302 """
303 raise NotImplementedError(self.get)
305 def get_multivar(self, section: SectionLike, name: NameLike) -> Iterator[Value]:
306 """Retrieve the contents of a multivar configuration setting.
308 Args:
309 section: Tuple with section name and optional subsection namee
310 name: Variable name
311 Returns:
312 Contents of the setting as iterable
313 Raises:
314 KeyError: if the value is not set
315 """
316 raise NotImplementedError(self.get_multivar)
318 @overload
319 def get_boolean(
320 self, section: SectionLike, name: NameLike, default: bool
321 ) -> bool: ...
323 @overload
324 def get_boolean(self, section: SectionLike, name: NameLike) -> Optional[bool]: ...
326 def get_boolean(
327 self, section: SectionLike, name: NameLike, default: Optional[bool] = None
328 ) -> Optional[bool]:
329 """Retrieve a configuration setting as boolean.
331 Args:
332 section: Tuple with section name and optional subsection name
333 name: Name of the setting, including section and possible
334 subsection.
336 Returns:
337 Contents of the setting
338 """
339 try:
340 value = self.get(section, name)
341 except KeyError:
342 return default
343 if value.lower() == b"true":
344 return True
345 elif value.lower() == b"false":
346 return False
347 raise ValueError(f"not a valid boolean string: {value!r}")
349 def set(
350 self, section: SectionLike, name: NameLike, value: Union[ValueLike, bool]
351 ) -> None:
352 """Set a configuration value.
354 Args:
355 section: Tuple with section name and optional subsection namee
356 name: Name of the configuration value, including section
357 and optional subsection
358 value: value of the setting
359 """
360 raise NotImplementedError(self.set)
362 def items(self, section: SectionLike) -> Iterator[tuple[Name, Value]]:
363 """Iterate over the configuration pairs for a specific section.
365 Args:
366 section: Tuple with section name and optional subsection namee
367 Returns:
368 Iterator over (name, value) pairs
369 """
370 raise NotImplementedError(self.items)
372 def sections(self) -> Iterator[Section]:
373 """Iterate over the sections.
375 Returns: Iterator over section tuples
376 """
377 raise NotImplementedError(self.sections)
379 def has_section(self, name: Section) -> bool:
380 """Check if a specified section exists.
382 Args:
383 name: Name of section to check for
384 Returns:
385 boolean indicating whether the section exists
386 """
387 return name in self.sections()
390class ConfigDict(Config):
391 """Git configuration stored in a dictionary."""
393 def __init__(
394 self,
395 values: Union[
396 MutableMapping[Section, MutableMapping[Name, Value]], None
397 ] = None,
398 encoding: Union[str, None] = None,
399 ) -> None:
400 """Create a new ConfigDict."""
401 if encoding is None:
402 encoding = sys.getdefaultencoding()
403 self.encoding = encoding
404 self._values: CaseInsensitiveOrderedMultiDict[
405 Section, CaseInsensitiveOrderedMultiDict[Name, Value]
406 ] = CaseInsensitiveOrderedMultiDict.make(
407 values, default_factory=CaseInsensitiveOrderedMultiDict
408 )
410 def __repr__(self) -> str:
411 return f"{self.__class__.__name__}({self._values!r})"
413 def __eq__(self, other: object) -> bool:
414 return isinstance(other, self.__class__) and other._values == self._values
416 def __getitem__(self, key: Section) -> CaseInsensitiveOrderedMultiDict[Name, Value]:
417 return self._values.__getitem__(key)
419 def __setitem__(self, key: Section, value: MutableMapping[Name, Value]) -> None:
420 return self._values.__setitem__(key, value)
422 def __delitem__(self, key: Section) -> None:
423 return self._values.__delitem__(key)
425 def __iter__(self) -> Iterator[Section]:
426 return self._values.__iter__()
428 def __len__(self) -> int:
429 return self._values.__len__()
431 def keys(self) -> KeysView[Section]:
432 return self._values.keys()
434 @classmethod
435 def _parse_setting(cls, name: str) -> tuple[str, Optional[str], str]:
436 parts = name.split(".")
437 if len(parts) == 3:
438 return (parts[0], parts[1], parts[2])
439 else:
440 return (parts[0], None, parts[1])
442 def _check_section_and_name(
443 self, section: SectionLike, name: NameLike
444 ) -> tuple[Section, Name]:
445 if not isinstance(section, tuple):
446 section = (section,)
448 checked_section = tuple(
449 [
450 subsection.encode(self.encoding)
451 if not isinstance(subsection, bytes)
452 else subsection
453 for subsection in section
454 ]
455 )
457 if not isinstance(name, bytes):
458 name = name.encode(self.encoding)
460 return checked_section, name
462 def get_multivar(self, section: SectionLike, name: NameLike) -> Iterator[Value]:
463 section, name = self._check_section_and_name(section, name)
465 if len(section) > 1:
466 try:
467 return self._values[section].get_all(name)
468 except KeyError:
469 pass
471 return self._values[(section[0],)].get_all(name)
473 def get(
474 self,
475 section: SectionLike,
476 name: NameLike,
477 ) -> Value:
478 section, name = self._check_section_and_name(section, name)
480 if len(section) > 1:
481 try:
482 return self._values[section][name]
483 except KeyError:
484 pass
486 return self._values[(section[0],)][name]
488 def set(
489 self,
490 section: SectionLike,
491 name: NameLike,
492 value: Union[ValueLike, bool],
493 ) -> None:
494 section, name = self._check_section_and_name(section, name)
496 if isinstance(value, bool):
497 value = b"true" if value else b"false"
499 if not isinstance(value, bytes):
500 value = value.encode(self.encoding)
502 section_dict = self._values.setdefault(section)
503 if hasattr(section_dict, "set"):
504 section_dict.set(name, value)
505 else:
506 section_dict[name] = value
508 def add(
509 self,
510 section: SectionLike,
511 name: NameLike,
512 value: Union[ValueLike, bool],
513 ) -> None:
514 """Add a value to a configuration setting, creating a multivar if needed."""
515 section, name = self._check_section_and_name(section, name)
517 if isinstance(value, bool):
518 value = b"true" if value else b"false"
520 if not isinstance(value, bytes):
521 value = value.encode(self.encoding)
523 self._values.setdefault(section)[name] = value
525 def items(self, section: SectionLike) -> Iterator[tuple[Name, Value]]:
526 section_bytes, _ = self._check_section_and_name(section, b"")
527 section_dict = self._values.get(section_bytes)
528 if section_dict is not None:
529 return iter(section_dict.items())
530 return iter([])
532 def sections(self) -> Iterator[Section]:
533 return iter(self._values.keys())
536def _format_string(value: bytes) -> bytes:
537 if (
538 value.startswith((b" ", b"\t"))
539 or value.endswith((b" ", b"\t"))
540 or b"#" in value
541 ):
542 return b'"' + _escape_value(value) + b'"'
543 else:
544 return _escape_value(value)
547_ESCAPE_TABLE = {
548 ord(b"\\"): ord(b"\\"),
549 ord(b'"'): ord(b'"'),
550 ord(b"n"): ord(b"\n"),
551 ord(b"t"): ord(b"\t"),
552 ord(b"b"): ord(b"\b"),
553}
554_COMMENT_CHARS = [ord(b"#"), ord(b";")]
555_WHITESPACE_CHARS = [ord(b"\t"), ord(b" ")]
558def _parse_string(value: bytes) -> bytes:
559 value = bytearray(value.strip())
560 ret = bytearray()
561 whitespace = bytearray()
562 in_quotes = False
563 i = 0
564 while i < len(value):
565 c = value[i]
566 if c == ord(b"\\"):
567 i += 1
568 if i >= len(value):
569 # Backslash at end of string - treat as literal backslash
570 if whitespace:
571 ret.extend(whitespace)
572 whitespace = bytearray()
573 ret.append(ord(b"\\"))
574 else:
575 try:
576 v = _ESCAPE_TABLE[value[i]]
577 if whitespace:
578 ret.extend(whitespace)
579 whitespace = bytearray()
580 ret.append(v)
581 except KeyError:
582 # Unknown escape sequence - treat backslash as literal and process next char normally
583 if whitespace:
584 ret.extend(whitespace)
585 whitespace = bytearray()
586 ret.append(ord(b"\\"))
587 i -= 1 # Reprocess the character after the backslash
588 elif c == ord(b'"'):
589 in_quotes = not in_quotes
590 elif c in _COMMENT_CHARS and not in_quotes:
591 # the rest of the line is a comment
592 break
593 elif c in _WHITESPACE_CHARS:
594 whitespace.append(c)
595 else:
596 if whitespace:
597 ret.extend(whitespace)
598 whitespace = bytearray()
599 ret.append(c)
600 i += 1
602 if in_quotes:
603 raise ValueError("missing end quote")
605 return bytes(ret)
608def _escape_value(value: bytes) -> bytes:
609 """Escape a value."""
610 value = value.replace(b"\\", b"\\\\")
611 value = value.replace(b"\r", b"\\r")
612 value = value.replace(b"\n", b"\\n")
613 value = value.replace(b"\t", b"\\t")
614 value = value.replace(b'"', b'\\"')
615 return value
618def _check_variable_name(name: bytes) -> bool:
619 for i in range(len(name)):
620 c = name[i : i + 1]
621 if not c.isalnum() and c != b"-":
622 return False
623 return True
626def _check_section_name(name: bytes) -> bool:
627 for i in range(len(name)):
628 c = name[i : i + 1]
629 if not c.isalnum() and c not in (b"-", b"."):
630 return False
631 return True
634def _strip_comments(line: bytes) -> bytes:
635 comment_bytes = {ord(b"#"), ord(b";")}
636 quote = ord(b'"')
637 string_open = False
638 # Normalize line to bytearray for simple 2/3 compatibility
639 for i, character in enumerate(bytearray(line)):
640 # Comment characters outside balanced quotes denote comment start
641 if character == quote:
642 string_open = not string_open
643 elif not string_open and character in comment_bytes:
644 return line[:i]
645 return line
648def _is_line_continuation(value: bytes) -> bool:
649 """Check if a value ends with a line continuation backslash.
651 A line continuation occurs when a line ends with a backslash that is:
652 1. Not escaped (not preceded by another backslash)
653 2. Not within quotes
655 Args:
656 value: The value to check
658 Returns:
659 True if the value ends with a line continuation backslash
660 """
661 if not value.endswith((b"\\\n", b"\\\r\n")):
662 return False
664 # Remove only the newline characters, keep the content including the backslash
665 if value.endswith(b"\\\r\n"):
666 content = value[:-2] # Remove \r\n, keep the \
667 else:
668 content = value[:-1] # Remove \n, keep the \
670 if not content.endswith(b"\\"):
671 return False
673 # Count consecutive backslashes at the end
674 backslash_count = 0
675 for i in range(len(content) - 1, -1, -1):
676 if content[i : i + 1] == b"\\":
677 backslash_count += 1
678 else:
679 break
681 # If we have an odd number of backslashes, the last one is a line continuation
682 # If we have an even number, they are all escaped and there's no continuation
683 return backslash_count % 2 == 1
686def _parse_section_header_line(line: bytes) -> tuple[Section, bytes]:
687 # Parse section header ("[bla]")
688 line = _strip_comments(line).rstrip()
689 in_quotes = False
690 escaped = False
691 for i, c in enumerate(line):
692 if escaped:
693 escaped = False
694 continue
695 if c == ord(b'"'):
696 in_quotes = not in_quotes
697 if c == ord(b"\\"):
698 escaped = True
699 if c == ord(b"]") and not in_quotes:
700 last = i
701 break
702 else:
703 raise ValueError("expected trailing ]")
704 pts = line[1:last].split(b" ", 1)
705 line = line[last + 1 :]
706 section: Section
707 if len(pts) == 2:
708 # Handle subsections - Git allows more complex syntax for certain sections like includeIf
709 if pts[1][:1] == b'"' and pts[1][-1:] == b'"':
710 # Standard quoted subsection
711 pts[1] = pts[1][1:-1]
712 elif pts[0] == b"includeIf":
713 # Special handling for includeIf sections which can have complex conditions
714 # Git allows these without strict quote validation
715 pts[1] = pts[1].strip()
716 if pts[1][:1] == b'"' and pts[1][-1:] == b'"':
717 pts[1] = pts[1][1:-1]
718 else:
719 # Other sections must have quoted subsections
720 raise ValueError(f"Invalid subsection {pts[1]!r}")
721 if not _check_section_name(pts[0]):
722 raise ValueError(f"invalid section name {pts[0]!r}")
723 section = (pts[0], pts[1])
724 else:
725 if not _check_section_name(pts[0]):
726 raise ValueError(f"invalid section name {pts[0]!r}")
727 pts = pts[0].split(b".", 1)
728 if len(pts) == 2:
729 section = (pts[0], pts[1])
730 else:
731 section = (pts[0],)
732 return section, line
735class ConfigFile(ConfigDict):
736 """A Git configuration file, like .git/config or ~/.gitconfig."""
738 def __init__(
739 self,
740 values: Union[
741 MutableMapping[Section, MutableMapping[Name, Value]], None
742 ] = None,
743 encoding: Union[str, None] = None,
744 ) -> None:
745 super().__init__(values=values, encoding=encoding)
746 self.path: Optional[str] = None
747 self._included_paths: set[str] = set() # Track included files to prevent cycles
749 @classmethod
750 def from_file(
751 cls,
752 f: BinaryIO,
753 *,
754 config_dir: Optional[str] = None,
755 included_paths: Optional[set[str]] = None,
756 include_depth: int = 0,
757 max_include_depth: int = DEFAULT_MAX_INCLUDE_DEPTH,
758 file_opener: Optional[FileOpener] = None,
759 condition_matchers: Optional[dict[str, ConditionMatcher]] = None,
760 ) -> "ConfigFile":
761 """Read configuration from a file-like object.
763 Args:
764 f: File-like object to read from
765 config_dir: Directory containing the config file (for relative includes)
766 included_paths: Set of already included paths (to prevent cycles)
767 include_depth: Current include depth (to prevent infinite recursion)
768 max_include_depth: Maximum allowed include depth
769 file_opener: Optional callback to open included files
770 condition_matchers: Optional dict of condition matchers for includeIf
771 """
772 if include_depth > max_include_depth:
773 # Prevent excessive recursion
774 raise ValueError(f"Maximum include depth ({max_include_depth}) exceeded")
776 ret = cls()
777 if included_paths is not None:
778 ret._included_paths = included_paths.copy()
780 section: Optional[Section] = None
781 setting = None
782 continuation = None
783 for lineno, line in enumerate(f.readlines()):
784 if lineno == 0 and line.startswith(b"\xef\xbb\xbf"):
785 line = line[3:]
786 line = line.lstrip()
787 if setting is None:
788 if len(line) > 0 and line[:1] == b"[":
789 section, line = _parse_section_header_line(line)
790 ret._values.setdefault(section)
791 if _strip_comments(line).strip() == b"":
792 continue
793 if section is None:
794 raise ValueError(f"setting {line!r} without section")
795 try:
796 setting, value = line.split(b"=", 1)
797 except ValueError:
798 setting = line
799 value = b"true"
800 setting = setting.strip()
801 if not _check_variable_name(setting):
802 raise ValueError(f"invalid variable name {setting!r}")
803 if _is_line_continuation(value):
804 if value.endswith(b"\\\r\n"):
805 continuation = value[:-3]
806 else:
807 continuation = value[:-2]
808 else:
809 continuation = None
810 value = _parse_string(value)
811 ret._values[section][setting] = value
813 # Process include/includeIf directives
814 ret._handle_include_directive(
815 section,
816 setting,
817 value,
818 config_dir=config_dir,
819 include_depth=include_depth,
820 max_include_depth=max_include_depth,
821 file_opener=file_opener,
822 condition_matchers=condition_matchers,
823 )
825 setting = None
826 else: # continuation line
827 assert continuation is not None
828 if _is_line_continuation(line):
829 if line.endswith(b"\\\r\n"):
830 continuation += line[:-3]
831 else:
832 continuation += line[:-2]
833 else:
834 continuation += line
835 value = _parse_string(continuation)
836 assert section is not None # Already checked above
837 ret._values[section][setting] = value
839 # Process include/includeIf directives
840 ret._handle_include_directive(
841 section,
842 setting,
843 value,
844 config_dir=config_dir,
845 include_depth=include_depth,
846 max_include_depth=max_include_depth,
847 file_opener=file_opener,
848 condition_matchers=condition_matchers,
849 )
851 continuation = None
852 setting = None
853 return ret
855 def _handle_include_directive(
856 self,
857 section: Optional[Section],
858 setting: bytes,
859 value: bytes,
860 *,
861 config_dir: Optional[str],
862 include_depth: int,
863 max_include_depth: int,
864 file_opener: Optional[FileOpener],
865 condition_matchers: Optional[dict[str, ConditionMatcher]],
866 ) -> None:
867 """Handle include/includeIf directives during config parsing."""
868 if (
869 section is not None
870 and setting == b"path"
871 and (
872 section[0].lower() == b"include"
873 or (len(section) > 1 and section[0].lower() == b"includeif")
874 )
875 ):
876 self._process_include(
877 section,
878 value,
879 config_dir=config_dir,
880 include_depth=include_depth,
881 max_include_depth=max_include_depth,
882 file_opener=file_opener,
883 condition_matchers=condition_matchers,
884 )
886 def _process_include(
887 self,
888 section: Section,
889 path_value: bytes,
890 *,
891 config_dir: Optional[str],
892 include_depth: int,
893 max_include_depth: int,
894 file_opener: Optional[FileOpener],
895 condition_matchers: Optional[dict[str, ConditionMatcher]],
896 ) -> None:
897 """Process an include or includeIf directive."""
898 path_str = path_value.decode(self.encoding, errors="replace")
900 # Handle includeIf conditions
901 if len(section) > 1 and section[0].lower() == b"includeif":
902 condition = section[1].decode(self.encoding, errors="replace")
903 if not self._evaluate_includeif_condition(
904 condition, config_dir, condition_matchers
905 ):
906 return
908 # Resolve the include path
909 include_path = self._resolve_include_path(path_str, config_dir)
910 if not include_path:
911 return
913 # Check for circular includes
914 try:
915 abs_path = str(Path(include_path).resolve())
916 except (OSError, ValueError) as e:
917 # Invalid path - log and skip
918 logger.debug("Invalid include path %r: %s", include_path, e)
919 return
920 if abs_path in self._included_paths:
921 return
923 # Load and merge the included file
924 try:
925 # Use provided file opener or default to GitFile
926 if file_opener is None:
928 def opener(path):
929 return GitFile(path, "rb")
930 else:
931 opener = file_opener
933 f = opener(include_path)
934 except (OSError, ValueError) as e:
935 # Git silently ignores missing or unreadable include files
936 # Log for debugging purposes
937 logger.debug("Invalid include path %r: %s", include_path, e)
938 else:
939 with f as included_file:
940 # Track this path to prevent cycles
941 self._included_paths.add(abs_path)
943 # Parse the included file
944 included_config = ConfigFile.from_file(
945 included_file,
946 config_dir=os.path.dirname(include_path),
947 included_paths=self._included_paths,
948 include_depth=include_depth + 1,
949 max_include_depth=max_include_depth,
950 file_opener=file_opener,
951 condition_matchers=condition_matchers,
952 )
954 # Merge the included configuration
955 self._merge_config(included_config)
957 def _merge_config(self, other: "ConfigFile") -> None:
958 """Merge another config file into this one."""
959 for section, values in other._values.items():
960 if section not in self._values:
961 self._values[section] = CaseInsensitiveOrderedMultiDict()
962 for key, value in values.items():
963 self._values[section][key] = value
965 def _resolve_include_path(
966 self, path: str, config_dir: Optional[str]
967 ) -> Optional[str]:
968 """Resolve an include path to an absolute path."""
969 # Expand ~ to home directory
970 path = os.path.expanduser(path)
972 # If path is relative and we have a config directory, make it relative to that
973 if not os.path.isabs(path) and config_dir:
974 path = os.path.join(config_dir, path)
976 return path
978 def _evaluate_includeif_condition(
979 self,
980 condition: str,
981 config_dir: Optional[str] = None,
982 condition_matchers: Optional[dict[str, ConditionMatcher]] = None,
983 ) -> bool:
984 """Evaluate an includeIf condition."""
985 # Try custom matchers first if provided
986 if condition_matchers:
987 for prefix, matcher in condition_matchers.items():
988 if condition.startswith(prefix):
989 return matcher(condition[len(prefix) :])
991 # Fall back to built-in matchers
992 if condition.startswith("hasconfig:"):
993 return self._evaluate_hasconfig_condition(condition[10:])
994 else:
995 # Unknown condition type - log and ignore (Git behavior)
996 logger.debug("Unknown includeIf condition: %r", condition)
997 return False
999 def _evaluate_hasconfig_condition(self, condition: str) -> bool:
1000 """Evaluate a hasconfig condition.
1002 Format: hasconfig:config.key:pattern
1003 Example: hasconfig:remote.*.url:ssh://org-*@github.com/**
1004 """
1005 # Split on the first colon to separate config key from pattern
1006 parts = condition.split(":", 1)
1007 if len(parts) != 2:
1008 logger.debug("Invalid hasconfig condition format: %r", condition)
1009 return False
1011 config_key, pattern = parts
1013 # Parse the config key to get section and name
1014 key_parts = config_key.split(".", 2)
1015 if len(key_parts) < 2:
1016 logger.debug("Invalid hasconfig config key: %r", config_key)
1017 return False
1019 # Handle wildcards in section names (e.g., remote.*)
1020 if len(key_parts) == 3 and key_parts[1] == "*":
1021 # Match any subsection
1022 section_prefix = key_parts[0].encode(self.encoding)
1023 name = key_parts[2].encode(self.encoding)
1025 # Check all sections that match the pattern
1026 for section in self.sections():
1027 if len(section) == 2 and section[0] == section_prefix:
1028 try:
1029 values = list(self.get_multivar(section, name))
1030 for value in values:
1031 if self._match_hasconfig_pattern(value, pattern):
1032 return True
1033 except KeyError:
1034 continue
1035 else:
1036 # Direct section lookup
1037 if len(key_parts) == 2:
1038 section = (key_parts[0].encode(self.encoding),)
1039 name = key_parts[1].encode(self.encoding)
1040 else:
1041 section = (
1042 key_parts[0].encode(self.encoding),
1043 key_parts[1].encode(self.encoding),
1044 )
1045 name = key_parts[2].encode(self.encoding)
1047 try:
1048 values = list(self.get_multivar(section, name))
1049 for value in values:
1050 if self._match_hasconfig_pattern(value, pattern):
1051 return True
1052 except KeyError:
1053 pass
1055 return False
1057 def _match_hasconfig_pattern(self, value: bytes, pattern: str) -> bool:
1058 """Match a config value against a hasconfig pattern.
1060 Supports simple glob patterns like ``*`` and ``**``.
1061 """
1062 value_str = value.decode(self.encoding, errors="replace")
1063 return match_glob_pattern(value_str, pattern)
1065 @classmethod
1066 def from_path(
1067 cls,
1068 path: Union[str, os.PathLike],
1069 *,
1070 max_include_depth: int = DEFAULT_MAX_INCLUDE_DEPTH,
1071 file_opener: Optional[FileOpener] = None,
1072 condition_matchers: Optional[dict[str, ConditionMatcher]] = None,
1073 ) -> "ConfigFile":
1074 """Read configuration from a file on disk.
1076 Args:
1077 path: Path to the configuration file
1078 max_include_depth: Maximum allowed include depth
1079 file_opener: Optional callback to open included files
1080 condition_matchers: Optional dict of condition matchers for includeIf
1081 """
1082 abs_path = os.fspath(path)
1083 config_dir = os.path.dirname(abs_path)
1085 # Use provided file opener or default to GitFile
1086 if file_opener is None:
1088 def opener(p):
1089 return GitFile(p, "rb")
1090 else:
1091 opener = file_opener
1093 with opener(abs_path) as f:
1094 ret = cls.from_file(
1095 f,
1096 config_dir=config_dir,
1097 max_include_depth=max_include_depth,
1098 file_opener=file_opener,
1099 condition_matchers=condition_matchers,
1100 )
1101 ret.path = abs_path
1102 return ret
1104 def write_to_path(self, path: Optional[Union[str, os.PathLike]] = None) -> None:
1105 """Write configuration to a file on disk."""
1106 if path is None:
1107 if self.path is None:
1108 raise ValueError("No path specified and no default path available")
1109 path_to_use: Union[str, os.PathLike] = self.path
1110 else:
1111 path_to_use = path
1112 with GitFile(path_to_use, "wb") as f:
1113 self.write_to_file(f)
1115 def write_to_file(self, f: BinaryIO) -> None:
1116 """Write configuration to a file-like object."""
1117 for section, values in self._values.items():
1118 try:
1119 section_name, subsection_name = section
1120 except ValueError:
1121 (section_name,) = section
1122 subsection_name = None
1123 if subsection_name is None:
1124 f.write(b"[" + section_name + b"]\n")
1125 else:
1126 f.write(b"[" + section_name + b' "' + subsection_name + b'"]\n')
1127 for key, value in values.items():
1128 value = _format_string(value)
1129 f.write(b"\t" + key + b" = " + value + b"\n")
1132def get_xdg_config_home_path(*path_segments: str) -> str:
1133 xdg_config_home = os.environ.get(
1134 "XDG_CONFIG_HOME",
1135 os.path.expanduser("~/.config/"),
1136 )
1137 return os.path.join(xdg_config_home, *path_segments)
1140def _find_git_in_win_path() -> Iterator[str]:
1141 for exe in ("git.exe", "git.cmd"):
1142 for path in os.environ.get("PATH", "").split(";"):
1143 if os.path.exists(os.path.join(path, exe)):
1144 # in windows native shells (powershell/cmd) exe path is
1145 # .../Git/bin/git.exe or .../Git/cmd/git.exe
1146 #
1147 # in git-bash exe path is .../Git/mingw64/bin/git.exe
1148 git_dir, _bin_dir = os.path.split(path)
1149 yield git_dir
1150 parent_dir, basename = os.path.split(git_dir)
1151 if basename == "mingw32" or basename == "mingw64":
1152 yield parent_dir
1153 break
1156def _find_git_in_win_reg() -> Iterator[str]:
1157 import platform
1158 import winreg
1160 if platform.machine() == "AMD64":
1161 subkey = (
1162 "SOFTWARE\\Wow6432Node\\Microsoft\\Windows\\"
1163 "CurrentVersion\\Uninstall\\Git_is1"
1164 )
1165 else:
1166 subkey = "SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\Uninstall\\Git_is1"
1168 for key in (winreg.HKEY_CURRENT_USER, winreg.HKEY_LOCAL_MACHINE): # type: ignore
1169 with suppress(OSError):
1170 with winreg.OpenKey(key, subkey) as k: # type: ignore
1171 val, typ = winreg.QueryValueEx(k, "InstallLocation") # type: ignore
1172 if typ == winreg.REG_SZ: # type: ignore
1173 yield val
1176# There is no set standard for system config dirs on windows. We try the
1177# following:
1178# - %PROGRAMDATA%/Git/config - (deprecated) Windows config dir per CGit docs
1179# - %PROGRAMFILES%/Git/etc/gitconfig - Git for Windows (msysgit) config dir
1180# Used if CGit installation (Git/bin/git.exe) is found in PATH in the
1181# system registry
1182def get_win_system_paths() -> Iterator[str]:
1183 if "PROGRAMDATA" in os.environ:
1184 yield os.path.join(os.environ["PROGRAMDATA"], "Git", "config")
1186 for git_dir in _find_git_in_win_path():
1187 yield os.path.join(git_dir, "etc", "gitconfig")
1188 for git_dir in _find_git_in_win_reg():
1189 yield os.path.join(git_dir, "etc", "gitconfig")
1192class StackedConfig(Config):
1193 """Configuration which reads from multiple config files.."""
1195 def __init__(
1196 self, backends: list[ConfigFile], writable: Optional[ConfigFile] = None
1197 ) -> None:
1198 self.backends = backends
1199 self.writable = writable
1201 def __repr__(self) -> str:
1202 return f"<{self.__class__.__name__} for {self.backends!r}>"
1204 @classmethod
1205 def default(cls) -> "StackedConfig":
1206 return cls(cls.default_backends())
1208 @classmethod
1209 def default_backends(cls) -> list[ConfigFile]:
1210 """Retrieve the default configuration.
1212 See git-config(1) for details on the files searched.
1213 """
1214 paths = []
1215 paths.append(os.path.expanduser("~/.gitconfig"))
1216 paths.append(get_xdg_config_home_path("git", "config"))
1218 if "GIT_CONFIG_NOSYSTEM" not in os.environ:
1219 paths.append("/etc/gitconfig")
1220 if sys.platform == "win32":
1221 paths.extend(get_win_system_paths())
1223 backends = []
1224 for path in paths:
1225 try:
1226 cf = ConfigFile.from_path(path)
1227 except FileNotFoundError:
1228 continue
1229 backends.append(cf)
1230 return backends
1232 def get(self, section: SectionLike, name: NameLike) -> Value:
1233 if not isinstance(section, tuple):
1234 section = (section,)
1235 for backend in self.backends:
1236 try:
1237 return backend.get(section, name)
1238 except KeyError:
1239 pass
1240 raise KeyError(name)
1242 def get_multivar(self, section: SectionLike, name: NameLike) -> Iterator[Value]:
1243 if not isinstance(section, tuple):
1244 section = (section,)
1245 for backend in self.backends:
1246 try:
1247 yield from backend.get_multivar(section, name)
1248 except KeyError:
1249 pass
1251 def set(
1252 self, section: SectionLike, name: NameLike, value: Union[ValueLike, bool]
1253 ) -> None:
1254 if self.writable is None:
1255 raise NotImplementedError(self.set)
1256 return self.writable.set(section, name, value)
1258 def sections(self) -> Iterator[Section]:
1259 seen = set()
1260 for backend in self.backends:
1261 for section in backend.sections():
1262 if section not in seen:
1263 seen.add(section)
1264 yield section
1267def read_submodules(
1268 path: Union[str, os.PathLike],
1269) -> Iterator[tuple[bytes, bytes, bytes]]:
1270 """Read a .gitmodules file."""
1271 cfg = ConfigFile.from_path(path)
1272 return parse_submodules(cfg)
1275def parse_submodules(config: ConfigFile) -> Iterator[tuple[bytes, bytes, bytes]]:
1276 """Parse a gitmodules GitConfig file, returning submodules.
1278 Args:
1279 config: A `ConfigFile`
1280 Returns:
1281 list of tuples (submodule path, url, name),
1282 where name is quoted part of the section's name.
1283 """
1284 for section in config.sections():
1285 section_kind, section_name = section
1286 if section_kind == b"submodule":
1287 try:
1288 sm_path = config.get(section, b"path")
1289 sm_url = config.get(section, b"url")
1290 yield (sm_path, sm_url, section_name)
1291 except KeyError:
1292 # If either path or url is missing, just ignore this
1293 # submodule entry and move on to the next one. This is
1294 # how git itself handles malformed .gitmodule entries.
1295 pass
1298def iter_instead_of(config: Config, push: bool = False) -> Iterable[tuple[str, str]]:
1299 """Iterate over insteadOf / pushInsteadOf values."""
1300 for section in config.sections():
1301 if section[0] != b"url":
1302 continue
1303 replacement = section[1]
1304 try:
1305 needles = list(config.get_multivar(section, "insteadOf"))
1306 except KeyError:
1307 needles = []
1308 if push:
1309 try:
1310 needles += list(config.get_multivar(section, "pushInsteadOf"))
1311 except KeyError:
1312 pass
1313 for needle in needles:
1314 assert isinstance(needle, bytes)
1315 yield needle.decode("utf-8"), replacement.decode("utf-8")
1318def apply_instead_of(config: Config, orig_url: str, push: bool = False) -> str:
1319 """Apply insteadOf / pushInsteadOf to a URL."""
1320 longest_needle = ""
1321 updated_url = orig_url
1322 for needle, replacement in iter_instead_of(config, push):
1323 if not orig_url.startswith(needle):
1324 continue
1325 if len(longest_needle) < len(needle):
1326 longest_needle = needle
1327 updated_url = replacement + orig_url[len(needle) :]
1328 return updated_url