# Copyright (C) 2008, 2009 Michael Trier (mtrier@gmail.com) and contributors
#
# This module is part of GitPython and is released under the
# 3-Clause BSD License: https://opensource.org/license/bsd-3-clause/
"""Parser for reading and writing configuration files."""
__all__ = ["GitConfigParser", "SectionConstraint"]
import abc
import configparser as cp
import fnmatch
from functools import wraps
import inspect
from io import BufferedReader, IOBase
import logging
import os
import os.path as osp
import re
import sys
from git.compat import defenc, force_text
from git.util import LockFile
# typing-------------------------------------------------------
from typing import (
Any,
Callable,
Generic,
IO,
List,
Dict,
Sequence,
TYPE_CHECKING,
Tuple,
TypeVar,
Union,
cast,
)
from git.types import Lit_config_levels, ConfigLevels_Tup, PathLike, assert_never, _T
if TYPE_CHECKING:
from io import BytesIO
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]
# -------------------------------------------------------------
_logger = logging.getLogger(__name__)
CONFIG_LEVELS: ConfigLevels_Tup = ("system", "user", "global", "repository")
"""The configuration level of a configuration file."""
CONDITIONAL_INCLUDE_REGEXP = re.compile(r"(? Union[bool, int]:
return self._merge_includes and len(self._included_paths())
def _included_paths(self) -> List[Tuple[str, str]]:
"""List all paths that must be included to configuration.
:return:
The list of paths, where each path is a tuple of (option, value).
"""
def _all_items(section: str) -> List[Tuple[str, str]]:
"""Return all (key, value) pairs for a section, including duplicate keys."""
return [
(key, value)
for key, values in self._sections[section].items_all()
if key != "__name__"
for value in values
]
paths = []
for section in self.sections():
if section == "include":
paths += _all_items(section)
match = CONDITIONAL_INCLUDE_REGEXP.search(section)
if match is None or self._repo is None:
continue
keyword = match.group(1)
value = match.group(2).strip()
if keyword in ["gitdir", "gitdir/i"]:
value = osp.expanduser(value)
git_dir = os.fspath(self._repo.git_dir) if self._repo.git_dir else None
if sys.platform == "win32":
git_dir = git_dir.replace("\\", "/") if git_dir else None
drive, _tail = osp.splitdrive(value)
if not drive and not any(value.startswith(s) for s in ["./", "/"]):
value = "**/" + value
if value.endswith("/"):
value += "**"
# Ensure that glob is always case insensitive if required.
if keyword.endswith("/i"):
value = re.sub(
r"[a-zA-Z]",
lambda m: f"[{m.group().lower()!r}{m.group().upper()!r}]",
value,
)
if git_dir and fnmatch.fnmatchcase(git_dir, value):
paths += _all_items(section)
elif keyword == "onbranch":
try:
branch_name = self._repo.active_branch.name
except TypeError:
# Ignore section if active branch cannot be retrieved.
continue
if fnmatch.fnmatchcase(branch_name, value):
paths += _all_items(section)
elif keyword == "hasconfig:remote.*.url":
for remote in self._repo.remotes:
if fnmatch.fnmatchcase(remote.url, value):
paths += _all_items(section)
break
return paths
def read(self) -> None: # type: ignore[override]
"""Read the data stored in the files we have been initialized with.
This will ignore files that cannot be read, possibly leaving an empty
configuration.
:raise IOError:
If a file cannot be handled.
"""
if self._is_initialized:
return
self._is_initialized = True
files_to_read: List[Union[PathLike, IO]] = [""]
if isinstance(self._file_or_files, (str, os.PathLike)):
# For str or Path, as str is a type of Sequence.
files_to_read = [self._file_or_files]
elif not isinstance(self._file_or_files, (tuple, list, Sequence)):
# Could merge with above isinstance once runtime type known.
files_to_read = [self._file_or_files]
else: # For lists or tuples.
files_to_read = list(self._file_or_files)
# END ensure we have a copy of the paths to handle
files_to_read = [osp.abspath(path) if isinstance(path, (str, os.PathLike)) else path for path in files_to_read]
seen = set(files_to_read)
num_read_include_files = 0
while files_to_read:
file_path = files_to_read.pop(0)
file_ok = False
if hasattr(file_path, "seek"):
# Must be a file-object.
# TODO: Replace cast with assert to narrow type, once sure.
file_path = cast(IO[bytes], file_path)
self._read(file_path, file_path.name)
else:
try:
with open(file_path, "rb") as fp:
file_ok = True
self._read(fp, fp.name)
except IOError:
continue
# Read includes and append those that we didn't handle yet. We expect all
# paths to be normalized and absolute (and will ensure that is the case).
if self._has_includes():
for _, include_path in self._included_paths():
if include_path.startswith("~"):
include_path = osp.expanduser(include_path)
if not osp.isabs(include_path):
if not file_ok:
continue
# END ignore relative paths if we don't know the configuration file path
file_path = cast(PathLike, file_path)
assert osp.isabs(file_path), "Need absolute paths to be sure our cycle checks will work"
include_path = osp.join(osp.dirname(file_path), include_path)
# END make include path absolute
include_path = osp.normpath(include_path)
if include_path in seen or not os.access(include_path, os.R_OK):
continue
seen.add(include_path)
# Insert included file to the top to be considered first.
files_to_read.insert(0, include_path)
num_read_include_files += 1
# END each include path in configuration file
# END handle includes
# END for each file object to read
# If there was no file included, we can safely write back (potentially) the
# configuration file without altering its meaning.
if num_read_include_files == 0:
self._merge_includes = False
def _write(self, fp: IO) -> None:
"""Write an .ini-format representation of the configuration state in
git compatible format."""
def write_section(name: str, section_dict: _OMD) -> None:
fp.write(("[%s]\n" % name).encode(defenc))
values: Sequence[str] # Runtime only gets str in tests, but should be whatever _OMD stores.
v: str
for key, values in section_dict.items_all():
if key == "__name__":
continue
for v in values:
value = self._value_to_string(v)
if any(char in value for char in '\n\t\b\\"#;') or value[:1].isspace() or value[-1:].isspace():
value = value.replace("\\", "\\\\").replace('"', '\\"')
value = '"%s\\\n"' % value.replace("\n", "\\n").replace("\t", "\\t").replace("\b", "\\b")
fp.write(("\t%s = %s\n" % (key, value)).encode(defenc))
# END if key is not __name__
# END section writing
if self._defaults:
write_section(cp.DEFAULTSECT, self._defaults)
value: _OMD
for name, value in self._sections.items():
write_section(name, value)
def items(self, section_name: str) -> List[Tuple[str, str]]: # type: ignore[override]
""":return: list((option, value), ...) pairs of all items in the given section"""
return [(k, v) for k, v in super().items(section_name) if k != "__name__"]
def items_all(self, section_name: str) -> List[Tuple[str, List[str]]]:
""":return: list((option, [values...]), ...) pairs of all items in the given section"""
rv = _OMD(self._defaults)
for k, vs in self._sections[section_name].items_all():
if k == "__name__":
continue
if k in rv and rv.getall(k) == vs:
continue
for v in vs:
rv.add(k, v)
return rv.items_all()
@needs_values
def write(self) -> None:
"""Write changes to our file, if there are changes at all.
:raise IOError:
If this is a read-only writer instance or if we could not obtain a file
lock.
"""
self._assure_writable("write")
if not self._dirty:
return
if isinstance(self._file_or_files, (list, tuple)):
raise AssertionError(
"Cannot write back if there is not exactly a single file to write to, have %i files"
% len(self._file_or_files)
)
# END assert multiple files
if self._has_includes():
_logger.debug(
"Skipping write-back of configuration file as include files were merged in."
+ "Set merge_includes=False to prevent this."
)
return
# END stop if we have include files
sections: List[_OMD] = [self._defaults]
section: _OMD
stored_section: _OMD
values: List[Any]
raw_value: Any
for _, stored_section in self._sections.items():
sections.append(stored_section)
for section in sections:
for key, values in section.items_all():
if key != "__name__":
for raw_value in values:
if "\r" in self._value_to_string(raw_value) or "\x00" in self._value_to_string(raw_value):
raise ValueError("Git config values must not contain CR or NUL")
fp = self._file_or_files
# We have a physical file on disk, so get a lock.
is_file_lock = isinstance(fp, (str, os.PathLike, IOBase)) # TODO: Use PathLike (having dropped 3.5).
if is_file_lock and self._lock is not None: # Else raise error?
self._lock._obtain_lock()
if not hasattr(fp, "seek"):
fp = cast(PathLike, fp)
with open(fp, "wb") as fp_open:
self._write(fp_open)
else:
fp = cast("BytesIO", fp)
fp.seek(0)
# Make sure we do not overwrite into an existing file.
if hasattr(fp, "truncate"):
fp.truncate()
self._write(fp)
def _assure_writable(self, method_name: str) -> None:
if self.read_only:
raise IOError("Cannot execute non-constant method %s.%s" % (self, method_name))
def add_section(self, section: "cp._SectionName") -> None:
"""Assures added options will stay in order."""
self._assure_config_name_safe(section, "section")
return super().add_section(section)
@property
def read_only(self) -> bool:
""":return: ``True`` if this instance may change the configuration file"""
return self._read_only
# FIXME: Figure out if default or return type can really include bool.
def get_value(
self,
section: str,
option: str,
default: Union[int, float, str, bool, None] = None,
) -> Union[int, float, str, bool]:
"""Get an option's value.
If multiple values are specified for this option in the section, the last one
specified is returned.
:param default:
If not ``None``, the given default value will be returned in case the option
did not exist.
:return:
A properly typed value, either int, float or string
:raise TypeError:
In case the value could not be understood.
Otherwise the exceptions known to the ConfigParser will be raised.
"""
try:
valuestr = self.get(section, option)
except Exception:
if default is not None:
return default
raise
return self._string_to_value(valuestr)
def get_values(
self,
section: str,
option: str,
default: Union[int, float, str, bool, None] = None,
) -> List[Union[int, float, str, bool]]:
"""Get an option's values.
If multiple values are specified for this option in the section, all are
returned.
:param default:
If not ``None``, a list containing the given default value will be returned
in case the option did not exist.
:return:
A list of properly typed values, either int, float or string
:raise TypeError:
In case the value could not be understood.
Otherwise the exceptions known to the ConfigParser will be raised.
"""
try:
self.sections()
lst = self._sections[section].getall(option)
except Exception:
if default is not None:
return [default]
raise
return [self._string_to_value(valuestr) for valuestr in lst]
def _string_to_value(self, valuestr: str) -> Union[int, float, str, bool]:
types = (int, float)
for numtype in types:
try:
val = numtype(valuestr)
# truncated value ?
if val != float(valuestr):
continue
return val
except (ValueError, TypeError):
continue
# END for each numeric type
# Try boolean values as git uses them.
vl = valuestr.lower()
if vl == "false":
return False
if vl == "true":
return True
if not isinstance(valuestr, str):
raise TypeError(
"Invalid value type: only int, long, float and str are allowed",
valuestr,
)
return valuestr
def _value_to_string(self, value: Union[str, bytes, int, float, bool]) -> str:
if isinstance(value, (int, float, bool)):
return str(value)
return force_text(value)
def _value_to_string_safe(self, value: Union[str, bytes, int, float, bool]) -> str:
value_str = self._value_to_string(value)
if UNSAFE_CONFIG_CHARS_RE.search(value_str):
raise ValueError("Git config values must not contain CR, LF, or NUL")
return value_str
def _assure_config_name_safe(self, name: "cp._SectionName", label: str) -> None:
if isinstance(name, str) and UNSAFE_CONFIG_CHARS_RE.search(name):
raise ValueError("Git config %s names must not contain CR, LF, or NUL" % label)
if label == "option" and isinstance(name, str) and not VALID_CONFIG_OPTION_NAME_RE.fullmatch(name):
raise ValueError("Git config option names may contain only letters, digits, '-', '_', or '.'")
if label == "section" and isinstance(name, str):
in_quotes = False
escaped = False
for index, char in enumerate(name):
if escaped:
escaped = False
elif in_quotes and char == "\\":
escaped = True
elif char == '"':
if not in_quotes and (index == 0 or name[index - 1] not in " \t"):
raise ValueError("Git config quoted subsection names must begin after whitespace")
in_quotes = not in_quotes
elif char == "]" and not in_quotes:
raise ValueError("Git config section names must not contain an unquoted closing bracket")
if in_quotes:
raise ValueError("Git config section names must not contain an unterminated quote")
@needs_values
@set_dirty_and_flush_changes
def set(
self,
section: str,
option: str,
value: Union[str, bytes, int, float, bool, None] = None,
) -> None:
self._assure_config_name_safe(section, "section")
self._assure_config_name_safe(option, "option")
if value is not None:
value = self._value_to_string_safe(value)
return super().set(section, option, value)
@needs_values
@set_dirty_and_flush_changes
def set_value(self, section: str, option: str, value: Union[str, bytes, int, float, bool]) -> "GitConfigParser":
"""Set the given option in section to the given value.
This will create the section if required, and will not throw as opposed to the
default ConfigParser ``set`` method.
:param section:
Name of the section in which the option resides or should reside.
:param option:
Name of the options whose value to set.
:param value:
Value to set the option to. It must be a string or convertible to a string.
:return:
This instance
"""
self._assure_config_name_safe(section, "section")
self._assure_config_name_safe(option, "option")
value_str = self._value_to_string_safe(value)
if not self.has_section(section):
self.add_section(section)
super().set(section, option, value_str)
return self
@needs_values
@set_dirty_and_flush_changes
def add_value(self, section: str, option: str, value: Union[str, bytes, int, float, bool]) -> "GitConfigParser":
"""Add a value for the given option in section.
This will create the section if required, and will not throw as opposed to the
default ConfigParser ``set`` method. The value becomes the new value of the
option as returned by :meth:`get_value`, and appends to the list of values
returned by :meth:`get_values`.
:param section:
Name of the section in which the option resides or should reside.
:param option:
Name of the option.
:param value:
Value to add to option. It must be a string or convertible to a string.
:return:
This instance
"""
self._assure_config_name_safe(section, "section")
self._assure_config_name_safe(option, "option")
value_str = self._value_to_string_safe(value)
if not self.has_section(section):
self.add_section(section)
self._sections[section].add(option, value_str)
return self
def rename_section(self, section: str, new_name: str) -> "GitConfigParser":
"""Rename the given section to `new_name`.
:raise ValueError:
If:
* `section` doesn't exist.
* A section with `new_name` does already exist.
:return:
This instance
"""
if not self.has_section(section):
raise ValueError("Source section '%s' doesn't exist" % section)
self._assure_config_name_safe(new_name, "section")
if self.has_section(new_name):
raise ValueError("Destination section '%s' already exists" % new_name)
super().add_section(new_name)
new_section = self._sections[new_name]
for k, vs in self.items_all(section):
new_section.setall(k, vs)
# END for each value to copy
# This call writes back the changes, which is why we don't have the respective
# decorator.
self.remove_section(section)
return self