diff --git a/git/config.py b/git/config.py index 821710ea3..8f94d1d9d 100644 --- a/git/config.py +++ b/git/config.py @@ -7,11 +7,18 @@ __all__ = ["GitConfigParser", "SectionConstraint"] -import abc -import configparser as cp +from collections import OrderedDict +from collections.abc import MutableMapping +from configparser import ( + DuplicateSectionError, + Error as ConfigError, + MissingSectionHeaderError, + NoOptionError, + NoSectionError, + ParsingError, +) import fnmatch from functools import wraps -import inspect from io import BufferedReader, IOBase import logging import os @@ -29,8 +36,10 @@ Callable, Generic, IO, + Iterator, List, Dict, + NoReturn, Sequence, TYPE_CHECKING, Tuple, @@ -47,17 +56,6 @@ from git.repo.base import Repo T_ConfigParser = TypeVar("T_ConfigParser", bound="GitConfigParser") -T_OMD_value = TypeVar("T_OMD_value", str, bytes, int, float, bool) - -if sys.version_info[:3] < (3, 7, 2): - # typing.Ordereddict not added until Python 3.7.2. - from collections import OrderedDict - - OrderedDict_OMD = OrderedDict -else: - from typing import OrderedDict - - OrderedDict_OMD = OrderedDict[str, List[T_OMD_value]] # type: ignore[assignment, misc] # ------------------------------------------------------------- @@ -66,45 +64,225 @@ CONFIG_LEVELS: ConfigLevels_Tup = ("system", "user", "global", "repository") """The configuration level of a configuration file.""" -CONDITIONAL_INCLUDE_REGEXP = re.compile(r"(?<=includeIf )\"(gitdir|gitdir/i|onbranch|hasconfig:remote\.\*\.url):(.+)\"") +CONDITIONAL_INCLUDE_REGEXP = re.compile( + r'(?<=includeif )"(gitdir|gitdir/i|onbranch|hasconfig:remote\.\*\.url):(.+)"', + re.IGNORECASE, +) """Section pattern to detect conditional includes. See: https://git-scm.com/docs/git-config#_conditional_includes """ UNSAFE_CONFIG_CHARS_RE = re.compile(r"[\r\n\x00]") -"""Characters that cannot be safely written in config names or values.""" +"""Characters that cannot be safely written in config names.""" +UNSAFE_CONFIG_VALUE_CHARS_RE = re.compile(r"\x00") +"""Characters that cannot be represented in Git config values.""" -class MetaParserBuilder(abc.ABCMeta): # noqa: B024 - """Utility class wrapping base-class methods into decorators that assure read-only - properties.""" +_MISSING = object() - def __new__(cls, name: str, bases: Tuple, clsdict: Dict[str, Any]) -> "MetaParserBuilder": - """Equip all base-class methods with a needs_values decorator, and all non-const - methods with a :func:`set_dirty_and_flush_changes` decorator in addition to - that. - """ - kmm = "_mutating_methods_" - if kmm in clsdict: - mutating_methods = clsdict[kmm] - for base in bases: - methods = (t for t in inspect.getmembers(base, inspect.isroutine) if not t[0].startswith("_")) - for method_name, method in methods: - if method_name in clsdict: - continue - method_with_values = needs_values(method) - if method_name in mutating_methods: - method_with_values = set_dirty_and_flush_changes(method_with_values) - # END mutating methods handling - clsdict[method_name] = method_with_values - # END for each name/method pair - # END for each base - # END if mutating methods configuration is set +def _escape_section_subsection(value: str) -> str: + """Return *value* escaped for Git's double-quoted subsection syntax.""" + return value.replace("\\", "\\\\").replace('"', '\\"') + - new_type = super().__new__(cls, name, bases, clsdict) - return new_type +def _escape_config_value(value: str) -> str: + """Return *value* in a canonical representation accepted by Git.""" + escaped = value.replace("\\", "\\\\") + escaped = escaped.replace('"', '\\"') + escaped = escaped.replace("\n", "\\n").replace("\t", "\\t").replace("\b", "\\b") + return '"%s"' % escaped + + +class _GitConfigFileParser: + """Parse Git config syntax using the grammar in Git's ``config.c``. + + The state machine mirrors Git's ``git_parse_source()``, ``get_base_var()``, + ``get_value()``, and ``parse_value()``. It intentionally deals only with syntax; + includes are resolved by :class:`GitConfigParser` after each file is parsed. + """ + + def __init__(self, data: str, source: str) -> None: + self._data = data[1:] if data.startswith("\ufeff") else data + self._source = source + self._position = 0 + self._line_number = 1 + self._current_section: Union[str, None] = None + + @staticmethod + def _is_key_char(char: str) -> bool: + return char.isascii() and (char.isalnum() or char == "-") + + @staticmethod + def _is_space(char: str) -> bool: + return char in " \t\n\v\f\r" + + def _get_char(self) -> Union[str, None]: + if self._position == len(self._data): + return None + + char = self._data[self._position] + self._position += 1 + if char == "\r" and self._position < len(self._data) and self._data[self._position] == "\n": + self._position += 1 + char = "\n" + if char == "\n": + self._line_number += 1 + return char + + def _error(self, line_number: Union[int, None] = None) -> NoReturn: + number = self._line_number if line_number is None else line_number + lines = self._data.splitlines() + line = lines[number - 1] if 0 < number <= len(lines) else "" + error = ParsingError(self._source) + error.append(number, repr(line)) + raise error + + def _parse_section(self, line_number: int) -> str: + base: List[str] = [] + while True: + char = self._get_char() + if char is None or char == "\n": + self._error(line_number) + if char == "]": + if not base: + self._error(line_number) + return "".join(base) + if self._is_space(char): + break + if not self._is_key_char(char) and char != ".": + self._error(line_number) + base.append(char) + + while char is not None and self._is_space(char) and char != "\n": + char = self._get_char() + if char != '"' or not base: + self._error(line_number) + + subsection: List[str] = [] + while True: + char = self._get_char() + if char is None or char == "\n": + self._error(line_number) + if char == '"': + break + if char == "\\": + char = self._get_char() + if char is None or char == "\n": + self._error(line_number) + subsection.append(char) + + if self._get_char() != "]": + self._error(line_number) + return '%s "%s"' % ("".join(base), _escape_section_subsection("".join(subsection))) + + def _parse_value(self, line_number: int) -> str: + value: List[str] = [] + pending_whitespace: List[str] = [] + quoted = False + comment = False + + while True: + char = self._get_char() + if char is None or char == "\n": + if quoted: + self._error(line_number) + return "".join(value) + if char == "\x00": + self._error(line_number) + if comment: + continue + if self._is_space(char) and not quoted: + if value: + pending_whitespace.append(char) + continue + if not quoted and char in ";#": + comment = True + continue + + if pending_whitespace: + value.extend(pending_whitespace) + pending_whitespace = [] + if char == "\\": + char = self._get_char() + if char is None: + self._error(line_number) + if char == "\n": + continue + escapes = {"t": "\t", "b": "\b", "n": "\n", "\\": "\\", '"': '"'} + if char not in escapes: + self._error(line_number) + value.append(escapes[char]) + elif char == '"': + quoted = not quoted + else: + value.append(char) + + def _parse_entry(self, first_char: str, line_number: int) -> Tuple[str, Union[str, None]]: + option = [first_char] + char = self._get_char() + while char is not None and self._is_key_char(char): + option.append(char) + char = self._get_char() + while char in (" ", "\t"): + char = self._get_char() + + # Git distinguishes a valueless key from a key whose value is "true". + if char is None or char == "\n": + return "".join(option), None + if char != "=": + self._error(line_number) + return "".join(option), self._parse_value(line_number) + + def parse(self) -> List[Tuple[str, Union[str, None], Union[str, None]]]: + events: List[Tuple[str, Union[str, None], Union[str, None]]] = [] + comment = False + + while True: + line_number = self._line_number + char = self._get_char() + if char is None: + return events + if char == "\n": + comment = False + continue + if comment: + continue + if self._is_space(char): + continue + if char in "#;": + comment = True + continue + if char == "[": + self._current_section = self._parse_section(line_number) + events.append((self._current_section, None, None)) + continue + if not (char.isascii() and char.isalpha()): + self._error(line_number) + if self._current_section is None: + line = self._data.splitlines()[line_number - 1] + raise MissingSectionHeaderError(self._source, line_number, line) + option, value = self._parse_entry(char, line_number) + events.append((self._current_section, option, value)) + + +def _canonical_section_name(name: str) -> str: + """Validate and canonicalize one section name using Git's own grammar.""" + events = _GitConfigFileParser("[%s]" % name, "
").parse() + sections = [section for section, option, _ in events if option is None] + if len(sections) != 1: + raise ValueError("Git config section name does not identify exactly one section") + return sections[0] + + +def _section_name_key(name: str) -> str: + """Return Git's case-insensitive lookup key for a canonical section name.""" + canonical_name = _canonical_section_name(name) + subsection_start = canonical_name.find(' "') + if subsection_start == -1: + return canonical_name.lower() + return canonical_name[:subsection_start].lower() + canonical_name[subsection_start:] def needs_values(func: Callable[..., _T]) -> Callable[..., _T]: @@ -139,13 +317,12 @@ def flush_changes(self: "GitConfigParser", *args: Any, **kwargs: Any) -> _T: class SectionConstraint(Generic[T_ConfigParser]): - """Constrains a ConfigParser to only option commands which are constrained to - always use the section we have been initialized with. + """Constrain a configuration parser to commands for one section. - It supports all ConfigParser methods that operate on an option. + It supports all :class:`GitConfigParser` methods that operate on an option. :note: - If used as a context manager, will release the wrapped ConfigParser. + If used as a context manager, this releases the wrapped parser. """ __slots__ = ("_config", "_section_name") @@ -186,7 +363,7 @@ def _call_config(self, method: str, *args: Any, **kwargs: Any) -> Any: @property def config(self) -> T_ConfigParser: - """return: ConfigParser instance we constrain""" + """Return the :class:`GitConfigParser` instance being constrained.""" return self._config def release(self) -> None: @@ -202,49 +379,94 @@ def __exit__(self, exception_type: str, exception_value: str, traceback: str) -> self._config.__exit__(exception_type, exception_value, traceback) -class _OMD(OrderedDict_OMD): - """Ordered multi-dict.""" +class _GitConfigSectionData: + """Ordered, multi-valued storage for one Git configuration section.""" - def __setitem__(self, key: str, value: _T) -> None: - super().__setitem__(key, [value]) + def __init__(self) -> None: + self._values: "OrderedDict[str, List[Union[str, None]]]" = OrderedDict() - def add(self, key: str, value: Any) -> None: - if key not in self: - super().__setitem__(key, [value]) - return + def __contains__(self, option: str) -> bool: + return option in self._values - super().__getitem__(key).append(value) + def __len__(self) -> int: + return len(self._values) - def setall(self, key: str, values: List[_T]) -> None: - super().__setitem__(key, values) + def add(self, option: str, value: Union[str, None]) -> None: + self._values.setdefault(option, []).append(value) - def __getitem__(self, key: str) -> Any: - return super().__getitem__(key)[-1] + def set(self, option: str, value: Union[str, None]) -> None: + self._values[option] = [value] - def getlast(self, key: str) -> Any: - return super().__getitem__(key)[-1] + def setall(self, option: str, values: List[Union[str, None]]) -> None: + self._values[option] = list(values) + + def getlast(self, option: str) -> Union[str, None]: + return self._values[option][-1] + + def getall(self, option: str) -> List[Union[str, None]]: + return list(self._values[option]) + + def remove(self, option: str) -> bool: + if option not in self._values: + return False + del self._values[option] + return True + + def options(self) -> List[str]: + return list(self._values) + + def items_all(self) -> List[Tuple[str, List[Union[str, None]]]]: + return [(option, list(values)) for option, values in self._values.items()] - def setlast(self, key: str, value: Any) -> None: - if key not in self: - super().__setitem__(key, [value]) - return - prior = super().__getitem__(key) - prior[-1] = value +class _GitConfigSection(MutableMapping): + """Mapping-style view of one section in a :class:`GitConfigParser`.""" - def get(self, key: str, default: Union[_T, None] = None) -> Union[_T, None]: - return super().get(key, [default])[-1] + def __init__(self, parser: "GitConfigParser", name: str) -> None: + self._parser = parser + self._name = name - def getall(self, key: str) -> List[_T]: - return super().__getitem__(key) + def __repr__(self) -> str: + return "" % self._name - def items(self) -> List[Tuple[str, _T]]: # type: ignore[override] - """List of (key, last value for key).""" - return [(k, self[k]) for k in self] + def __getitem__(self, option: str) -> str: + return self._parser.get(self._name, option) - def items_all(self) -> List[Tuple[str, List[_T]]]: - """List of (key, list of values for key).""" - return [(k, self.getall(k)) for k in self] + def __setitem__(self, option: str, value: Any) -> None: + self._parser.set(self._name, option, value) + + def __delitem__(self, option: str) -> None: + if not self._parser.remove_option(self._name, option): + raise KeyError(option) + + def __iter__(self) -> Iterator[str]: + return iter(self._parser.options(self._name)) + + def __len__(self) -> int: + return len(self._parser.options(self._name)) + + def __contains__(self, option: object) -> bool: + return isinstance(option, str) and self._parser.has_option(self._name, option) + + def get(self, option: str, fallback: Any = None, **kwargs: Any) -> Any: # type: ignore[override] + return self._parser.get(self._name, option, fallback=fallback, **kwargs) + + def getint(self, option: str, fallback: Any = None, **kwargs: Any) -> Any: + return self._parser.getint(self._name, option, fallback=fallback, **kwargs) + + def getfloat(self, option: str, fallback: Any = None, **kwargs: Any) -> Any: + return self._parser.getfloat(self._name, option, fallback=fallback, **kwargs) + + def getboolean(self, option: str, fallback: Any = None, **kwargs: Any) -> Any: + return self._parser.getboolean(self._name, option, fallback=fallback, **kwargs) + + @property + def parser(self) -> "GitConfigParser": + return self._parser + + @property + def name(self) -> str: + return self._name def get_config_path(config_level: Lit_config_levels) -> str: @@ -271,7 +493,7 @@ def get_config_path(config_level: Lit_config_levels) -> str: ) -class GitConfigParser(cp.RawConfigParser, metaclass=MetaParserBuilder): +class GitConfigParser: """Implements specifics required to read git style configuration files. This variation behaves much like the :manpage:`git-config(1)` command, such that the @@ -285,8 +507,8 @@ class GitConfigParser(cp.RawConfigParser, metaclass=MetaParserBuilder): other instances to write concurrently. :note: - The config is case-sensitive even when queried, hence section and option names - must match perfectly. + Section and option names are case-insensitive, while subsection names are + case-sensitive, matching Git. :note: If used as a context manager, this will release the locked file. @@ -300,21 +522,6 @@ class GitConfigParser(cp.RawConfigParser, metaclass=MetaParserBuilder): A suitable alternative would be the :class:`~git.util.BlockingLockFile`. """ - re_comment = re.compile(r"^\s*[#;]") - # } END configuration - - optvalueonly_source = r"\s*(?P