12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328 |
- # config.py - Reading and writing Git config files
- # Copyright (C) 2011-2013 Jelmer Vernooij <jelmer@jelmer.uk>
- #
- # SPDX-License-Identifier: Apache-2.0 OR GPL-2.0-or-later
- # Dulwich is dual-licensed under the Apache License, Version 2.0 and the GNU
- # General Public License as public by the Free Software Foundation; version 2.0
- # or (at your option) any later version. You can redistribute it and/or
- # modify it under the terms of either of these two licenses.
- #
- # Unless required by applicable law or agreed to in writing, software
- # distributed under the License is distributed on an "AS IS" BASIS,
- # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- # See the License for the specific language governing permissions and
- # limitations under the License.
- #
- # You should have received a copy of the licenses; if not, see
- # <http://www.gnu.org/licenses/> for a copy of the GNU General Public License
- # and <http://www.apache.org/licenses/LICENSE-2.0> for a copy of the Apache
- # License, Version 2.0.
- #
- """Reading and writing Git configuration files.
- Todo:
- * preserve formatting when updating configuration files
- """
- import logging
- import os
- import re
- import sys
- from collections.abc import (
- ItemsView,
- Iterable,
- Iterator,
- KeysView,
- MutableMapping,
- ValuesView,
- )
- from contextlib import suppress
- from pathlib import Path
- from typing import (
- Any,
- BinaryIO,
- Callable,
- Generic,
- Optional,
- TypeVar,
- Union,
- overload,
- )
- from .file import GitFile
- ConfigKey = Union[str, bytes, tuple[Union[str, bytes], ...]]
- ConfigValue = Union[str, bytes, bool, int]
- logger = logging.getLogger(__name__)
- # Type for file opener callback
- FileOpener = Callable[[Union[str, os.PathLike]], BinaryIO]
- # Type for includeIf condition matcher
- # Takes the condition value (e.g., "main" for onbranch:main) and returns bool
- ConditionMatcher = Callable[[str], bool]
- # Security limits for include files
- MAX_INCLUDE_FILE_SIZE = 1024 * 1024 # 1MB max for included config files
- DEFAULT_MAX_INCLUDE_DEPTH = 10 # Maximum recursion depth for includes
- def _match_gitdir_pattern(
- path: bytes, pattern: bytes, ignorecase: bool = False
- ) -> bool:
- """Simple gitdir pattern matching for includeIf conditions.
- This handles the basic gitdir patterns used in includeIf directives.
- """
- # Convert to strings for easier manipulation
- path_str = path.decode("utf-8", errors="replace")
- pattern_str = pattern.decode("utf-8", errors="replace")
- # Normalize paths to use forward slashes for consistent matching
- path_str = path_str.replace("\\", "/")
- pattern_str = pattern_str.replace("\\", "/")
- if ignorecase:
- path_str = path_str.lower()
- pattern_str = pattern_str.lower()
- # Handle the common cases for gitdir patterns
- if pattern_str.startswith("**/") and pattern_str.endswith("/**"):
- # Pattern like **/dirname/** should match any path containing dirname
- dirname = pattern_str[3:-3] # Remove **/ and /**
- # Check if path contains the directory name as a path component
- return ("/" + dirname + "/") in path_str or path_str.endswith("/" + dirname)
- elif pattern_str.startswith("**/"):
- # Pattern like **/filename
- suffix = pattern_str[3:] # Remove **/
- return suffix in path_str or path_str.endswith("/" + suffix)
- elif pattern_str.endswith("/**"):
- # Pattern like /path/to/dir/** should match /path/to/dir and any subdirectory
- base_pattern = pattern_str[:-3] # Remove /**
- return path_str == base_pattern or path_str.startswith(base_pattern + "/")
- elif "**" in pattern_str:
- # Handle patterns with ** in the middle
- parts = pattern_str.split("**")
- if len(parts) == 2:
- prefix, suffix = parts
- # Path must start with prefix and end with suffix (if any)
- if prefix and not path_str.startswith(prefix):
- return False
- if suffix and not path_str.endswith(suffix):
- return False
- return True
- # Direct match or simple glob pattern
- if "*" in pattern_str or "?" in pattern_str or "[" in pattern_str:
- import fnmatch
- return fnmatch.fnmatch(path_str, pattern_str)
- else:
- return path_str == pattern_str
- def match_glob_pattern(value: str, pattern: str) -> bool:
- r"""Match a value against a glob pattern.
- Supports simple glob patterns like ``*`` and ``**``.
- Raises:
- ValueError: If the pattern is invalid
- """
- # Convert glob pattern to regex
- pattern_escaped = re.escape(pattern)
- # Replace escaped \*\* with .* (match anything)
- pattern_escaped = pattern_escaped.replace(r"\*\*", ".*")
- # Replace escaped \* with [^/]* (match anything except /)
- pattern_escaped = pattern_escaped.replace(r"\*", "[^/]*")
- # Anchor the pattern
- pattern_regex = f"^{pattern_escaped}$"
- try:
- return bool(re.match(pattern_regex, value))
- except re.error as e:
- raise ValueError(f"Invalid glob pattern {pattern!r}: {e}")
- def lower_key(key: ConfigKey) -> ConfigKey:
- if isinstance(key, (bytes, str)):
- return key.lower()
- if isinstance(key, tuple):
- # For config sections, only lowercase the section name (first element)
- # but preserve the case of subsection names (remaining elements)
- if len(key) > 0:
- first = key[0]
- assert isinstance(first, (bytes, str))
- return (first.lower(), *key[1:])
- return key
- raise TypeError(key)
- K = TypeVar("K", bound=ConfigKey) # Key type must be ConfigKey
- V = TypeVar("V") # Value type
- _T = TypeVar("_T") # For get() default parameter
- class CaseInsensitiveOrderedMultiDict(MutableMapping[K, V], Generic[K, V]):
- def __init__(self, default_factory: Optional[Callable[[], V]] = None) -> None:
- self._real: list[tuple[K, V]] = []
- self._keyed: dict[Any, V] = {}
- self._default_factory = default_factory
- @classmethod
- def make(
- cls, dict_in=None, default_factory=None
- ) -> "CaseInsensitiveOrderedMultiDict[K, V]":
- if isinstance(dict_in, cls):
- return dict_in
- out = cls(default_factory=default_factory)
- if dict_in is None:
- return out
- if not isinstance(dict_in, MutableMapping):
- raise TypeError
- for key, value in dict_in.items():
- out[key] = value
- return out
- def __len__(self) -> int:
- return len(self._keyed)
- def keys(self) -> KeysView[K]:
- return self._keyed.keys() # type: ignore[return-value]
- def items(self) -> ItemsView[K, V]:
- # Return a view that iterates over the real list to preserve order
- class OrderedItemsView(ItemsView[K, V]):
- def __init__(self, mapping: CaseInsensitiveOrderedMultiDict[K, V]):
- self._mapping = mapping
- def __iter__(self) -> Iterator[tuple[K, V]]:
- return iter(self._mapping._real)
- def __len__(self) -> int:
- return len(self._mapping._real)
- def __contains__(self, item: object) -> bool:
- if not isinstance(item, tuple) or len(item) != 2:
- return False
- key, value = item
- return any(k == key and v == value for k, v in self._mapping._real)
- return OrderedItemsView(self)
- def __iter__(self) -> Iterator[K]:
- return iter(self._keyed)
- def values(self) -> ValuesView[V]:
- return self._keyed.values()
- def __setitem__(self, key, value) -> None:
- self._real.append((key, value))
- self._keyed[lower_key(key)] = value
- def set(self, key, value) -> None:
- # This method replaces all existing values for the key
- lower = lower_key(key)
- self._real = [(k, v) for k, v in self._real if lower_key(k) != lower]
- self._real.append((key, value))
- self._keyed[lower] = value
- def __delitem__(self, key) -> None:
- key = lower_key(key)
- del self._keyed[key]
- for i, (actual, unused_value) in reversed(list(enumerate(self._real))):
- if lower_key(actual) == key:
- del self._real[i]
- def __getitem__(self, item: K) -> V:
- return self._keyed[lower_key(item)]
- def get(self, key: K, /, default: Union[V, _T, None] = None) -> Union[V, _T, None]: # type: ignore[override]
- try:
- return self[key]
- except KeyError:
- if default is not None:
- return default
- elif self._default_factory is not None:
- return self._default_factory()
- else:
- return None
- def get_all(self, key: K) -> Iterator[V]:
- lowered_key = lower_key(key)
- for actual, value in self._real:
- if lower_key(actual) == lowered_key:
- yield value
- def setdefault(self, key: K, default: Optional[V] = None) -> V:
- try:
- return self[key]
- except KeyError:
- if default is not None:
- self[key] = default
- return default
- elif self._default_factory is not None:
- value = self._default_factory()
- self[key] = value
- return value
- else:
- raise
- Name = bytes
- NameLike = Union[bytes, str]
- Section = tuple[bytes, ...]
- SectionLike = Union[bytes, str, tuple[Union[bytes, str], ...]]
- Value = bytes
- ValueLike = Union[bytes, str]
- class Config:
- """A Git configuration."""
- def get(self, section: SectionLike, name: NameLike) -> Value:
- """Retrieve the contents of a configuration setting.
- Args:
- section: Tuple with section name and optional subsection name
- name: Variable name
- Returns:
- Contents of the setting
- Raises:
- KeyError: if the value is not set
- """
- raise NotImplementedError(self.get)
- def get_multivar(self, section: SectionLike, name: NameLike) -> Iterator[Value]:
- """Retrieve the contents of a multivar configuration setting.
- Args:
- section: Tuple with section name and optional subsection namee
- name: Variable name
- Returns:
- Contents of the setting as iterable
- Raises:
- KeyError: if the value is not set
- """
- raise NotImplementedError(self.get_multivar)
- @overload
- def get_boolean(
- self, section: SectionLike, name: NameLike, default: bool
- ) -> bool: ...
- @overload
- def get_boolean(self, section: SectionLike, name: NameLike) -> Optional[bool]: ...
- def get_boolean(
- self, section: SectionLike, name: NameLike, default: Optional[bool] = None
- ) -> Optional[bool]:
- """Retrieve a configuration setting as boolean.
- Args:
- section: Tuple with section name and optional subsection name
- name: Name of the setting, including section and possible
- subsection.
- Returns:
- Contents of the setting
- """
- try:
- value = self.get(section, name)
- except KeyError:
- return default
- if value.lower() == b"true":
- return True
- elif value.lower() == b"false":
- return False
- raise ValueError(f"not a valid boolean string: {value!r}")
- def set(
- self, section: SectionLike, name: NameLike, value: Union[ValueLike, bool]
- ) -> None:
- """Set a configuration value.
- Args:
- section: Tuple with section name and optional subsection namee
- name: Name of the configuration value, including section
- and optional subsection
- value: value of the setting
- """
- raise NotImplementedError(self.set)
- def items(self, section: SectionLike) -> Iterator[tuple[Name, Value]]:
- """Iterate over the configuration pairs for a specific section.
- Args:
- section: Tuple with section name and optional subsection namee
- Returns:
- Iterator over (name, value) pairs
- """
- raise NotImplementedError(self.items)
- def sections(self) -> Iterator[Section]:
- """Iterate over the sections.
- Returns: Iterator over section tuples
- """
- raise NotImplementedError(self.sections)
- def has_section(self, name: Section) -> bool:
- """Check if a specified section exists.
- Args:
- name: Name of section to check for
- Returns:
- boolean indicating whether the section exists
- """
- return name in self.sections()
- class ConfigDict(Config):
- """Git configuration stored in a dictionary."""
- def __init__(
- self,
- values: Union[
- MutableMapping[Section, MutableMapping[Name, Value]], None
- ] = None,
- encoding: Union[str, None] = None,
- ) -> None:
- """Create a new ConfigDict."""
- if encoding is None:
- encoding = sys.getdefaultencoding()
- self.encoding = encoding
- self._values: CaseInsensitiveOrderedMultiDict[
- Section, CaseInsensitiveOrderedMultiDict[Name, Value]
- ] = CaseInsensitiveOrderedMultiDict.make(
- values, default_factory=CaseInsensitiveOrderedMultiDict
- )
- def __repr__(self) -> str:
- return f"{self.__class__.__name__}({self._values!r})"
- def __eq__(self, other: object) -> bool:
- return isinstance(other, self.__class__) and other._values == self._values
- def __getitem__(self, key: Section) -> CaseInsensitiveOrderedMultiDict[Name, Value]:
- return self._values.__getitem__(key)
- def __setitem__(self, key: Section, value: MutableMapping[Name, Value]) -> None:
- return self._values.__setitem__(key, value)
- def __delitem__(self, key: Section) -> None:
- return self._values.__delitem__(key)
- def __iter__(self) -> Iterator[Section]:
- return self._values.__iter__()
- def __len__(self) -> int:
- return self._values.__len__()
- def keys(self) -> KeysView[Section]:
- return self._values.keys()
- @classmethod
- def _parse_setting(cls, name: str) -> tuple[str, Optional[str], str]:
- parts = name.split(".")
- if len(parts) == 3:
- return (parts[0], parts[1], parts[2])
- else:
- return (parts[0], None, parts[1])
- def _check_section_and_name(
- self, section: SectionLike, name: NameLike
- ) -> tuple[Section, Name]:
- if not isinstance(section, tuple):
- section = (section,)
- checked_section = tuple(
- [
- subsection.encode(self.encoding)
- if not isinstance(subsection, bytes)
- else subsection
- for subsection in section
- ]
- )
- if not isinstance(name, bytes):
- name = name.encode(self.encoding)
- return checked_section, name
- def get_multivar(self, section: SectionLike, name: NameLike) -> Iterator[Value]:
- section, name = self._check_section_and_name(section, name)
- if len(section) > 1:
- try:
- return self._values[section].get_all(name)
- except KeyError:
- pass
- return self._values[(section[0],)].get_all(name)
- def get(
- self,
- section: SectionLike,
- name: NameLike,
- ) -> Value:
- section, name = self._check_section_and_name(section, name)
- if len(section) > 1:
- try:
- return self._values[section][name]
- except KeyError:
- pass
- return self._values[(section[0],)][name]
- def set(
- self,
- section: SectionLike,
- name: NameLike,
- value: Union[ValueLike, bool],
- ) -> None:
- section, name = self._check_section_and_name(section, name)
- if isinstance(value, bool):
- value = b"true" if value else b"false"
- if not isinstance(value, bytes):
- value = value.encode(self.encoding)
- section_dict = self._values.setdefault(section)
- if hasattr(section_dict, "set"):
- section_dict.set(name, value)
- else:
- section_dict[name] = value
- def add(
- self,
- section: SectionLike,
- name: NameLike,
- value: Union[ValueLike, bool],
- ) -> None:
- """Add a value to a configuration setting, creating a multivar if needed."""
- section, name = self._check_section_and_name(section, name)
- if isinstance(value, bool):
- value = b"true" if value else b"false"
- if not isinstance(value, bytes):
- value = value.encode(self.encoding)
- self._values.setdefault(section)[name] = value
- def items(self, section: SectionLike) -> Iterator[tuple[Name, Value]]:
- section_bytes, _ = self._check_section_and_name(section, b"")
- section_dict = self._values.get(section_bytes)
- if section_dict is not None:
- return iter(section_dict.items())
- return iter([])
- def sections(self) -> Iterator[Section]:
- return iter(self._values.keys())
- def _format_string(value: bytes) -> bytes:
- if (
- value.startswith((b" ", b"\t"))
- or value.endswith((b" ", b"\t"))
- or b"#" in value
- ):
- return b'"' + _escape_value(value) + b'"'
- else:
- return _escape_value(value)
- _ESCAPE_TABLE = {
- ord(b"\\"): ord(b"\\"),
- ord(b'"'): ord(b'"'),
- ord(b"n"): ord(b"\n"),
- ord(b"t"): ord(b"\t"),
- ord(b"b"): ord(b"\b"),
- }
- _COMMENT_CHARS = [ord(b"#"), ord(b";")]
- _WHITESPACE_CHARS = [ord(b"\t"), ord(b" ")]
- def _parse_string(value: bytes) -> bytes:
- value = bytearray(value.strip())
- ret = bytearray()
- whitespace = bytearray()
- in_quotes = False
- i = 0
- while i < len(value):
- c = value[i]
- if c == ord(b"\\"):
- i += 1
- if i >= len(value):
- # Backslash at end of string - treat as literal backslash
- if whitespace:
- ret.extend(whitespace)
- whitespace = bytearray()
- ret.append(ord(b"\\"))
- else:
- try:
- v = _ESCAPE_TABLE[value[i]]
- if whitespace:
- ret.extend(whitespace)
- whitespace = bytearray()
- ret.append(v)
- except KeyError:
- # Unknown escape sequence - treat backslash as literal and process next char normally
- if whitespace:
- ret.extend(whitespace)
- whitespace = bytearray()
- ret.append(ord(b"\\"))
- i -= 1 # Reprocess the character after the backslash
- elif c == ord(b'"'):
- in_quotes = not in_quotes
- elif c in _COMMENT_CHARS and not in_quotes:
- # the rest of the line is a comment
- break
- elif c in _WHITESPACE_CHARS:
- whitespace.append(c)
- else:
- if whitespace:
- ret.extend(whitespace)
- whitespace = bytearray()
- ret.append(c)
- i += 1
- if in_quotes:
- raise ValueError("missing end quote")
- return bytes(ret)
- def _escape_value(value: bytes) -> bytes:
- """Escape a value."""
- value = value.replace(b"\\", b"\\\\")
- value = value.replace(b"\r", b"\\r")
- value = value.replace(b"\n", b"\\n")
- value = value.replace(b"\t", b"\\t")
- value = value.replace(b'"', b'\\"')
- return value
- def _check_variable_name(name: bytes) -> bool:
- for i in range(len(name)):
- c = name[i : i + 1]
- if not c.isalnum() and c != b"-":
- return False
- return True
- def _check_section_name(name: bytes) -> bool:
- for i in range(len(name)):
- c = name[i : i + 1]
- if not c.isalnum() and c not in (b"-", b"."):
- return False
- return True
- def _strip_comments(line: bytes) -> bytes:
- comment_bytes = {ord(b"#"), ord(b";")}
- quote = ord(b'"')
- string_open = False
- # Normalize line to bytearray for simple 2/3 compatibility
- for i, character in enumerate(bytearray(line)):
- # Comment characters outside balanced quotes denote comment start
- if character == quote:
- string_open = not string_open
- elif not string_open and character in comment_bytes:
- return line[:i]
- return line
- def _is_line_continuation(value: bytes) -> bool:
- """Check if a value ends with a line continuation backslash.
- A line continuation occurs when a line ends with a backslash that is:
- 1. Not escaped (not preceded by another backslash)
- 2. Not within quotes
- Args:
- value: The value to check
- Returns:
- True if the value ends with a line continuation backslash
- """
- if not value.endswith((b"\\\n", b"\\\r\n")):
- return False
- # Remove only the newline characters, keep the content including the backslash
- if value.endswith(b"\\\r\n"):
- content = value[:-2] # Remove \r\n, keep the \
- else:
- content = value[:-1] # Remove \n, keep the \
- if not content.endswith(b"\\"):
- return False
- # Count consecutive backslashes at the end
- backslash_count = 0
- for i in range(len(content) - 1, -1, -1):
- if content[i : i + 1] == b"\\":
- backslash_count += 1
- else:
- break
- # If we have an odd number of backslashes, the last one is a line continuation
- # If we have an even number, they are all escaped and there's no continuation
- return backslash_count % 2 == 1
- def _parse_section_header_line(line: bytes) -> tuple[Section, bytes]:
- # Parse section header ("[bla]")
- line = _strip_comments(line).rstrip()
- in_quotes = False
- escaped = False
- for i, c in enumerate(line):
- if escaped:
- escaped = False
- continue
- if c == ord(b'"'):
- in_quotes = not in_quotes
- if c == ord(b"\\"):
- escaped = True
- if c == ord(b"]") and not in_quotes:
- last = i
- break
- else:
- raise ValueError("expected trailing ]")
- pts = line[1:last].split(b" ", 1)
- line = line[last + 1 :]
- section: Section
- if len(pts) == 2:
- # Handle subsections - Git allows more complex syntax for certain sections like includeIf
- if pts[1][:1] == b'"' and pts[1][-1:] == b'"':
- # Standard quoted subsection
- pts[1] = pts[1][1:-1]
- elif pts[0] == b"includeIf":
- # Special handling for includeIf sections which can have complex conditions
- # Git allows these without strict quote validation
- pts[1] = pts[1].strip()
- if pts[1][:1] == b'"' and pts[1][-1:] == b'"':
- pts[1] = pts[1][1:-1]
- else:
- # Other sections must have quoted subsections
- raise ValueError(f"Invalid subsection {pts[1]!r}")
- if not _check_section_name(pts[0]):
- raise ValueError(f"invalid section name {pts[0]!r}")
- section = (pts[0], pts[1])
- else:
- if not _check_section_name(pts[0]):
- raise ValueError(f"invalid section name {pts[0]!r}")
- pts = pts[0].split(b".", 1)
- if len(pts) == 2:
- section = (pts[0], pts[1])
- else:
- section = (pts[0],)
- return section, line
- class ConfigFile(ConfigDict):
- """A Git configuration file, like .git/config or ~/.gitconfig."""
- def __init__(
- self,
- values: Union[
- MutableMapping[Section, MutableMapping[Name, Value]], None
- ] = None,
- encoding: Union[str, None] = None,
- ) -> None:
- super().__init__(values=values, encoding=encoding)
- self.path: Optional[str] = None
- self._included_paths: set[str] = set() # Track included files to prevent cycles
- @classmethod
- def from_file(
- cls,
- f: BinaryIO,
- *,
- config_dir: Optional[str] = None,
- included_paths: Optional[set[str]] = None,
- include_depth: int = 0,
- max_include_depth: int = DEFAULT_MAX_INCLUDE_DEPTH,
- file_opener: Optional[FileOpener] = None,
- condition_matchers: Optional[dict[str, ConditionMatcher]] = None,
- ) -> "ConfigFile":
- """Read configuration from a file-like object.
- Args:
- f: File-like object to read from
- config_dir: Directory containing the config file (for relative includes)
- included_paths: Set of already included paths (to prevent cycles)
- include_depth: Current include depth (to prevent infinite recursion)
- max_include_depth: Maximum allowed include depth
- file_opener: Optional callback to open included files
- condition_matchers: Optional dict of condition matchers for includeIf
- """
- if include_depth > max_include_depth:
- # Prevent excessive recursion
- raise ValueError(f"Maximum include depth ({max_include_depth}) exceeded")
- ret = cls()
- if included_paths is not None:
- ret._included_paths = included_paths.copy()
- section: Optional[Section] = None
- setting = None
- continuation = None
- for lineno, line in enumerate(f.readlines()):
- if lineno == 0 and line.startswith(b"\xef\xbb\xbf"):
- line = line[3:]
- line = line.lstrip()
- if setting is None:
- if len(line) > 0 and line[:1] == b"[":
- section, line = _parse_section_header_line(line)
- ret._values.setdefault(section)
- if _strip_comments(line).strip() == b"":
- continue
- if section is None:
- raise ValueError(f"setting {line!r} without section")
- try:
- setting, value = line.split(b"=", 1)
- except ValueError:
- setting = line
- value = b"true"
- setting = setting.strip()
- if not _check_variable_name(setting):
- raise ValueError(f"invalid variable name {setting!r}")
- if _is_line_continuation(value):
- if value.endswith(b"\\\r\n"):
- continuation = value[:-3]
- else:
- continuation = value[:-2]
- else:
- continuation = None
- value = _parse_string(value)
- ret._values[section][setting] = value
- # Process include/includeIf directives
- ret._handle_include_directive(
- section,
- setting,
- value,
- config_dir=config_dir,
- include_depth=include_depth,
- max_include_depth=max_include_depth,
- file_opener=file_opener,
- condition_matchers=condition_matchers,
- )
- setting = None
- else: # continuation line
- assert continuation is not None
- if _is_line_continuation(line):
- if line.endswith(b"\\\r\n"):
- continuation += line[:-3]
- else:
- continuation += line[:-2]
- else:
- continuation += line
- value = _parse_string(continuation)
- assert section is not None # Already checked above
- ret._values[section][setting] = value
- # Process include/includeIf directives
- ret._handle_include_directive(
- section,
- setting,
- value,
- config_dir=config_dir,
- include_depth=include_depth,
- max_include_depth=max_include_depth,
- file_opener=file_opener,
- condition_matchers=condition_matchers,
- )
- continuation = None
- setting = None
- return ret
- def _handle_include_directive(
- self,
- section: Optional[Section],
- setting: bytes,
- value: bytes,
- *,
- config_dir: Optional[str],
- include_depth: int,
- max_include_depth: int,
- file_opener: Optional[FileOpener],
- condition_matchers: Optional[dict[str, ConditionMatcher]],
- ) -> None:
- """Handle include/includeIf directives during config parsing."""
- if (
- section is not None
- and setting == b"path"
- and (
- section[0].lower() == b"include"
- or (len(section) > 1 and section[0].lower() == b"includeif")
- )
- ):
- self._process_include(
- section,
- value,
- config_dir=config_dir,
- include_depth=include_depth,
- max_include_depth=max_include_depth,
- file_opener=file_opener,
- condition_matchers=condition_matchers,
- )
- def _process_include(
- self,
- section: Section,
- path_value: bytes,
- *,
- config_dir: Optional[str],
- include_depth: int,
- max_include_depth: int,
- file_opener: Optional[FileOpener],
- condition_matchers: Optional[dict[str, ConditionMatcher]],
- ) -> None:
- """Process an include or includeIf directive."""
- path_str = path_value.decode(self.encoding, errors="replace")
- # Handle includeIf conditions
- if len(section) > 1 and section[0].lower() == b"includeif":
- condition = section[1].decode(self.encoding, errors="replace")
- if not self._evaluate_includeif_condition(
- condition, config_dir, condition_matchers
- ):
- return
- # Resolve the include path
- include_path = self._resolve_include_path(path_str, config_dir)
- if not include_path:
- return
- # Check for circular includes
- try:
- abs_path = str(Path(include_path).resolve())
- except (OSError, ValueError) as e:
- # Invalid path - log and skip
- logger.debug("Invalid include path %r: %s", include_path, e)
- return
- if abs_path in self._included_paths:
- return
- # Load and merge the included file
- try:
- # Use provided file opener or default to GitFile
- if file_opener is None:
- def opener(path):
- return GitFile(path, "rb")
- else:
- opener = file_opener
- f = opener(include_path)
- except (OSError, ValueError) as e:
- # Git silently ignores missing or unreadable include files
- # Log for debugging purposes
- logger.debug("Invalid include path %r: %s", include_path, e)
- else:
- with f as included_file:
- # Track this path to prevent cycles
- self._included_paths.add(abs_path)
- # Parse the included file
- included_config = ConfigFile.from_file(
- included_file,
- config_dir=os.path.dirname(include_path),
- included_paths=self._included_paths,
- include_depth=include_depth + 1,
- max_include_depth=max_include_depth,
- file_opener=file_opener,
- condition_matchers=condition_matchers,
- )
- # Merge the included configuration
- self._merge_config(included_config)
- def _merge_config(self, other: "ConfigFile") -> None:
- """Merge another config file into this one."""
- for section, values in other._values.items():
- if section not in self._values:
- self._values[section] = CaseInsensitiveOrderedMultiDict()
- for key, value in values.items():
- self._values[section][key] = value
- def _resolve_include_path(
- self, path: str, config_dir: Optional[str]
- ) -> Optional[str]:
- """Resolve an include path to an absolute path."""
- # Expand ~ to home directory
- path = os.path.expanduser(path)
- # If path is relative and we have a config directory, make it relative to that
- if not os.path.isabs(path) and config_dir:
- path = os.path.join(config_dir, path)
- return path
- def _evaluate_includeif_condition(
- self,
- condition: str,
- config_dir: Optional[str] = None,
- condition_matchers: Optional[dict[str, ConditionMatcher]] = None,
- ) -> bool:
- """Evaluate an includeIf condition."""
- # Try custom matchers first if provided
- if condition_matchers:
- for prefix, matcher in condition_matchers.items():
- if condition.startswith(prefix):
- return matcher(condition[len(prefix) :])
- # Fall back to built-in matchers
- if condition.startswith("hasconfig:"):
- return self._evaluate_hasconfig_condition(condition[10:])
- else:
- # Unknown condition type - log and ignore (Git behavior)
- logger.debug("Unknown includeIf condition: %r", condition)
- return False
- def _evaluate_hasconfig_condition(self, condition: str) -> bool:
- """Evaluate a hasconfig condition.
- Format: hasconfig:config.key:pattern
- Example: hasconfig:remote.*.url:ssh://org-*@github.com/**
- """
- # Split on the first colon to separate config key from pattern
- parts = condition.split(":", 1)
- if len(parts) != 2:
- logger.debug("Invalid hasconfig condition format: %r", condition)
- return False
- config_key, pattern = parts
- # Parse the config key to get section and name
- key_parts = config_key.split(".", 2)
- if len(key_parts) < 2:
- logger.debug("Invalid hasconfig config key: %r", config_key)
- return False
- # Handle wildcards in section names (e.g., remote.*)
- if len(key_parts) == 3 and key_parts[1] == "*":
- # Match any subsection
- section_prefix = key_parts[0].encode(self.encoding)
- name = key_parts[2].encode(self.encoding)
- # Check all sections that match the pattern
- for section in self.sections():
- if len(section) == 2 and section[0] == section_prefix:
- try:
- values = list(self.get_multivar(section, name))
- for value in values:
- if self._match_hasconfig_pattern(value, pattern):
- return True
- except KeyError:
- continue
- else:
- # Direct section lookup
- if len(key_parts) == 2:
- section = (key_parts[0].encode(self.encoding),)
- name = key_parts[1].encode(self.encoding)
- else:
- section = (
- key_parts[0].encode(self.encoding),
- key_parts[1].encode(self.encoding),
- )
- name = key_parts[2].encode(self.encoding)
- try:
- values = list(self.get_multivar(section, name))
- for value in values:
- if self._match_hasconfig_pattern(value, pattern):
- return True
- except KeyError:
- pass
- return False
- def _match_hasconfig_pattern(self, value: bytes, pattern: str) -> bool:
- """Match a config value against a hasconfig pattern.
- Supports simple glob patterns like ``*`` and ``**``.
- """
- value_str = value.decode(self.encoding, errors="replace")
- return match_glob_pattern(value_str, pattern)
- @classmethod
- def from_path(
- cls,
- path: Union[str, os.PathLike],
- *,
- max_include_depth: int = DEFAULT_MAX_INCLUDE_DEPTH,
- file_opener: Optional[FileOpener] = None,
- condition_matchers: Optional[dict[str, ConditionMatcher]] = None,
- ) -> "ConfigFile":
- """Read configuration from a file on disk.
- Args:
- path: Path to the configuration file
- max_include_depth: Maximum allowed include depth
- file_opener: Optional callback to open included files
- condition_matchers: Optional dict of condition matchers for includeIf
- """
- abs_path = os.fspath(path)
- config_dir = os.path.dirname(abs_path)
- # Use provided file opener or default to GitFile
- if file_opener is None:
- def opener(p):
- return GitFile(p, "rb")
- else:
- opener = file_opener
- with opener(abs_path) as f:
- ret = cls.from_file(
- f,
- config_dir=config_dir,
- max_include_depth=max_include_depth,
- file_opener=file_opener,
- condition_matchers=condition_matchers,
- )
- ret.path = abs_path
- return ret
- def write_to_path(self, path: Optional[Union[str, os.PathLike]] = None) -> None:
- """Write configuration to a file on disk."""
- if path is None:
- if self.path is None:
- raise ValueError("No path specified and no default path available")
- path_to_use: Union[str, os.PathLike] = self.path
- else:
- path_to_use = path
- with GitFile(path_to_use, "wb") as f:
- self.write_to_file(f)
- def write_to_file(self, f: BinaryIO) -> None:
- """Write configuration to a file-like object."""
- for section, values in self._values.items():
- try:
- section_name, subsection_name = section
- except ValueError:
- (section_name,) = section
- subsection_name = None
- if subsection_name is None:
- f.write(b"[" + section_name + b"]\n")
- else:
- f.write(b"[" + section_name + b' "' + subsection_name + b'"]\n')
- for key, value in values.items():
- value = _format_string(value)
- f.write(b"\t" + key + b" = " + value + b"\n")
- def get_xdg_config_home_path(*path_segments: str) -> str:
- xdg_config_home = os.environ.get(
- "XDG_CONFIG_HOME",
- os.path.expanduser("~/.config/"),
- )
- return os.path.join(xdg_config_home, *path_segments)
- def _find_git_in_win_path() -> Iterator[str]:
- for exe in ("git.exe", "git.cmd"):
- for path in os.environ.get("PATH", "").split(";"):
- if os.path.exists(os.path.join(path, exe)):
- # in windows native shells (powershell/cmd) exe path is
- # .../Git/bin/git.exe or .../Git/cmd/git.exe
- #
- # in git-bash exe path is .../Git/mingw64/bin/git.exe
- git_dir, _bin_dir = os.path.split(path)
- yield git_dir
- parent_dir, basename = os.path.split(git_dir)
- if basename == "mingw32" or basename == "mingw64":
- yield parent_dir
- break
- def _find_git_in_win_reg() -> Iterator[str]:
- import platform
- import winreg
- if platform.machine() == "AMD64":
- subkey = (
- "SOFTWARE\\Wow6432Node\\Microsoft\\Windows\\"
- "CurrentVersion\\Uninstall\\Git_is1"
- )
- else:
- subkey = "SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\Uninstall\\Git_is1"
- for key in (winreg.HKEY_CURRENT_USER, winreg.HKEY_LOCAL_MACHINE): # type: ignore
- with suppress(OSError):
- with winreg.OpenKey(key, subkey) as k: # type: ignore
- val, typ = winreg.QueryValueEx(k, "InstallLocation") # type: ignore
- if typ == winreg.REG_SZ: # type: ignore
- yield val
- # There is no set standard for system config dirs on windows. We try the
- # following:
- # - %PROGRAMDATA%/Git/config - (deprecated) Windows config dir per CGit docs
- # - %PROGRAMFILES%/Git/etc/gitconfig - Git for Windows (msysgit) config dir
- # Used if CGit installation (Git/bin/git.exe) is found in PATH in the
- # system registry
- def get_win_system_paths() -> Iterator[str]:
- if "PROGRAMDATA" in os.environ:
- yield os.path.join(os.environ["PROGRAMDATA"], "Git", "config")
- for git_dir in _find_git_in_win_path():
- yield os.path.join(git_dir, "etc", "gitconfig")
- for git_dir in _find_git_in_win_reg():
- yield os.path.join(git_dir, "etc", "gitconfig")
- class StackedConfig(Config):
- """Configuration which reads from multiple config files.."""
- def __init__(
- self, backends: list[ConfigFile], writable: Optional[ConfigFile] = None
- ) -> None:
- self.backends = backends
- self.writable = writable
- def __repr__(self) -> str:
- return f"<{self.__class__.__name__} for {self.backends!r}>"
- @classmethod
- def default(cls) -> "StackedConfig":
- return cls(cls.default_backends())
- @classmethod
- def default_backends(cls) -> list[ConfigFile]:
- """Retrieve the default configuration.
- See git-config(1) for details on the files searched.
- """
- paths = []
- paths.append(os.path.expanduser("~/.gitconfig"))
- paths.append(get_xdg_config_home_path("git", "config"))
- if "GIT_CONFIG_NOSYSTEM" not in os.environ:
- paths.append("/etc/gitconfig")
- if sys.platform == "win32":
- paths.extend(get_win_system_paths())
- backends = []
- for path in paths:
- try:
- cf = ConfigFile.from_path(path)
- except FileNotFoundError:
- continue
- backends.append(cf)
- return backends
- def get(self, section: SectionLike, name: NameLike) -> Value:
- if not isinstance(section, tuple):
- section = (section,)
- for backend in self.backends:
- try:
- return backend.get(section, name)
- except KeyError:
- pass
- raise KeyError(name)
- def get_multivar(self, section: SectionLike, name: NameLike) -> Iterator[Value]:
- if not isinstance(section, tuple):
- section = (section,)
- for backend in self.backends:
- try:
- yield from backend.get_multivar(section, name)
- except KeyError:
- pass
- def set(
- self, section: SectionLike, name: NameLike, value: Union[ValueLike, bool]
- ) -> None:
- if self.writable is None:
- raise NotImplementedError(self.set)
- return self.writable.set(section, name, value)
- def sections(self) -> Iterator[Section]:
- seen = set()
- for backend in self.backends:
- for section in backend.sections():
- if section not in seen:
- seen.add(section)
- yield section
- def read_submodules(
- path: Union[str, os.PathLike],
- ) -> Iterator[tuple[bytes, bytes, bytes]]:
- """Read a .gitmodules file."""
- cfg = ConfigFile.from_path(path)
- return parse_submodules(cfg)
- def parse_submodules(config: ConfigFile) -> Iterator[tuple[bytes, bytes, bytes]]:
- """Parse a gitmodules GitConfig file, returning submodules.
- Args:
- config: A `ConfigFile`
- Returns:
- list of tuples (submodule path, url, name),
- where name is quoted part of the section's name.
- """
- for section in config.sections():
- section_kind, section_name = section
- if section_kind == b"submodule":
- try:
- sm_path = config.get(section, b"path")
- sm_url = config.get(section, b"url")
- yield (sm_path, sm_url, section_name)
- except KeyError:
- # If either path or url is missing, just ignore this
- # submodule entry and move on to the next one. This is
- # how git itself handles malformed .gitmodule entries.
- pass
- def iter_instead_of(config: Config, push: bool = False) -> Iterable[tuple[str, str]]:
- """Iterate over insteadOf / pushInsteadOf values."""
- for section in config.sections():
- if section[0] != b"url":
- continue
- replacement = section[1]
- try:
- needles = list(config.get_multivar(section, "insteadOf"))
- except KeyError:
- needles = []
- if push:
- try:
- needles += list(config.get_multivar(section, "pushInsteadOf"))
- except KeyError:
- pass
- for needle in needles:
- assert isinstance(needle, bytes)
- yield needle.decode("utf-8"), replacement.decode("utf-8")
- def apply_instead_of(config: Config, orig_url: str, push: bool = False) -> str:
- """Apply insteadOf / pushInsteadOf to a URL."""
- longest_needle = ""
- updated_url = orig_url
- for needle, replacement in iter_instead_of(config, push):
- if not orig_url.startswith(needle):
- continue
- if len(longest_needle) < len(needle):
- longest_needle = needle
- updated_url = replacement + orig_url[len(needle) :]
- return updated_url
|