Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/dulwich/config.py: 63%

1# config.py - Reading and writing Git config files

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.

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 Iterable, Iterator, KeysView, MutableMapping

33from contextlib import suppress

34from pathlib import Path

35from typing import (

36 Any,

37 BinaryIO,

38 Callable,

39 Optional,

40 Union,

41 overload,

42)

44from .file import GitFile

46logger = logging.getLogger(__name__)

48# Type for file opener callback

49FileOpener = Callable[[Union[str, os.PathLike]], BinaryIO]

51# Type for includeIf condition matcher

52# Takes the condition value (e.g., "main" for onbranch:main) and returns bool

53ConditionMatcher = Callable[[str], bool]

55# Security limits for include files

56MAX_INCLUDE_FILE_SIZE = 1024 * 1024 # 1MB max for included config files

57DEFAULT_MAX_INCLUDE_DEPTH = 10 # Maximum recursion depth for includes

59SENTINEL = object()

62def _match_gitdir_pattern(

63 path: bytes, pattern: bytes, ignorecase: bool = False

64) -> bool:

65 """Simple gitdir pattern matching for includeIf conditions.

67 This handles the basic gitdir patterns used in includeIf directives.

68 """

69 # Convert to strings for easier manipulation

70 path_str = path.decode("utf-8", errors="replace")

71 pattern_str = pattern.decode("utf-8", errors="replace")

73 # Normalize paths to use forward slashes for consistent matching

74 path_str = path_str.replace("\\", "/")

75 pattern_str = pattern_str.replace("\\", "/")

77 if ignorecase:

78 path_str = path_str.lower()

79 pattern_str = pattern_str.lower()

81 # Handle the common cases for gitdir patterns

82 if pattern_str.startswith("**/") and pattern_str.endswith("/**"):

83 # Pattern like **/dirname/** should match any path containing dirname

84 dirname = pattern_str[3:-3] # Remove **/ and /**

85 # Check if path contains the directory name as a path component

86 return ("/" + dirname + "/") in path_str or path_str.endswith("/" + dirname)

87 elif pattern_str.startswith("**/"):

88 # Pattern like **/filename

89 suffix = pattern_str[3:] # Remove **/

90 return suffix in path_str or path_str.endswith("/" + suffix)

91 elif pattern_str.endswith("/**"):

92 # Pattern like /path/to/dir/** should match /path/to/dir and any subdirectory

93 base_pattern = pattern_str[:-3] # Remove /**

94 return path_str == base_pattern or path_str.startswith(base_pattern + "/")

95 elif "**" in pattern_str:

96 # Handle patterns with ** in the middle

97 parts = pattern_str.split("**")

98 if len(parts) == 2:

99 prefix, suffix = parts

100 # Path must start with prefix and end with suffix (if any)

101 if prefix and not path_str.startswith(prefix):

102 return False

103 if suffix and not path_str.endswith(suffix):

104 return False

105 return True

106

107 # Direct match or simple glob pattern

108 if "*" in pattern_str or "?" in pattern_str or "[" in pattern_str:

109 import fnmatch

110

111 return fnmatch.fnmatch(path_str, pattern_str)

112 else:

113 return path_str == pattern_str

114

115

116def match_glob_pattern(value: str, pattern: str) -> bool:

117 r"""Match a value against a glob pattern.

118

119 Supports simple glob patterns like ``*`` and ``**``.

120

121 Raises:

122 ValueError: If the pattern is invalid

123 """

124 # Convert glob pattern to regex

125 pattern_escaped = re.escape(pattern)

126 # Replace escaped \*\* with .* (match anything)

127 pattern_escaped = pattern_escaped.replace(r"\*\*", ".*")

128 # Replace escaped \* with [^/]* (match anything except /)

129 pattern_escaped = pattern_escaped.replace(r"\*", "[^/]*")

130 # Anchor the pattern

131 pattern_regex = f"^{pattern_escaped}$"

132

133 try:

134 return bool(re.match(pattern_regex, value))

135 except re.error as e:

136 raise ValueError(f"Invalid glob pattern {pattern!r}: {e}")

137

138

139def lower_key(key):

140 if isinstance(key, (bytes, str)):

141 return key.lower()

142

143 if isinstance(key, Iterable):

144 # For config sections, only lowercase the section name (first element)

145 # but preserve the case of subsection names (remaining elements)

146 if len(key) > 0:

147 return (key[0].lower(), *key[1:])

148 return key

149

150 return key

151

152

153class CaseInsensitiveOrderedMultiDict(MutableMapping):

154 def __init__(self) -> None:

155 self._real: list[Any] = []

156 self._keyed: dict[Any, Any] = {}

157

158 @classmethod

159 def make(cls, dict_in=None):

160 if isinstance(dict_in, cls):

161 return dict_in

162

163 out = cls()

164

165 if dict_in is None:

166 return out

167

168 if not isinstance(dict_in, MutableMapping):

169 raise TypeError

170

171 for key, value in dict_in.items():

172 out[key] = value

173

174 return out

175

176 def __len__(self) -> int:

177 return len(self._keyed)

178

179 def keys(self) -> KeysView[tuple[bytes, ...]]:

180 return self._keyed.keys()

181

182 def items(self):

183 return iter(self._real)

184

185 def __iter__(self):

186 return self._keyed.__iter__()

187

188 def values(self):

189 return self._keyed.values()

190

191 def __setitem__(self, key, value) -> None:

192 self._real.append((key, value))

193 self._keyed[lower_key(key)] = value

194

195 def set(self, key, value) -> None:

196 # This method replaces all existing values for the key

197 lower = lower_key(key)

198 self._real = [(k, v) for k, v in self._real if lower_key(k) != lower]

199 self._real.append((key, value))

200 self._keyed[lower] = value

201

202 def __delitem__(self, key) -> None:

203 key = lower_key(key)

204 del self._keyed[key]

205 for i, (actual, unused_value) in reversed(list(enumerate(self._real))):

206 if lower_key(actual) == key:

207 del self._real[i]

208

209 def __getitem__(self, item):

210 return self._keyed[lower_key(item)]

211

212 def get(self, key, default=SENTINEL):

213 try:

214 return self[key]

215 except KeyError:

216 pass

217

218 if default is SENTINEL:

219 return type(self)()

220

221 return default

222

223 def get_all(self, key):

224 key = lower_key(key)

225 for actual, value in self._real:

226 if lower_key(actual) == key:

227 yield value

228

229 def setdefault(self, key, default=SENTINEL):

230 try:

231 return self[key]

232 except KeyError:

233 self[key] = self.get(key, default)

234

235 return self[key]

236

237

238Name = bytes

239NameLike = Union[bytes, str]

240Section = tuple[bytes, ...]

241SectionLike = Union[bytes, str, tuple[Union[bytes, str], ...]]

242Value = bytes

243ValueLike = Union[bytes, str]

244

245

246class Config:

247 """A Git configuration."""

248

249 def get(self, section: SectionLike, name: NameLike) -> Value:

250 """Retrieve the contents of a configuration setting.

251

252 Args:

253 section: Tuple with section name and optional subsection name

254 name: Variable name

255 Returns:

256 Contents of the setting

257 Raises:

258 KeyError: if the value is not set

259 """

260 raise NotImplementedError(self.get)

261

262 def get_multivar(self, section: SectionLike, name: NameLike) -> Iterator[Value]:

263 """Retrieve the contents of a multivar configuration setting.

264

265 Args:

266 section: Tuple with section name and optional subsection namee

267 name: Variable name

268 Returns:

269 Contents of the setting as iterable

270 Raises:

271 KeyError: if the value is not set

272 """

273 raise NotImplementedError(self.get_multivar)

274

275 @overload

276 def get_boolean(

277 self, section: SectionLike, name: NameLike, default: bool

278 ) -> bool: ...

279

280 @overload

281 def get_boolean(self, section: SectionLike, name: NameLike) -> Optional[bool]: ...

282

283 def get_boolean(

284 self, section: SectionLike, name: NameLike, default: Optional[bool] = None

285 ) -> Optional[bool]:

286 """Retrieve a configuration setting as boolean.

287

288 Args:

289 section: Tuple with section name and optional subsection name

290 name: Name of the setting, including section and possible

291 subsection.

292

293 Returns:

294 Contents of the setting

295 """

296 try:

297 value = self.get(section, name)

298 except KeyError:

299 return default

300 if value.lower() == b"true":

301 return True

302 elif value.lower() == b"false":

303 return False

304 raise ValueError(f"not a valid boolean string: {value!r}")

305

306 def set(

307 self, section: SectionLike, name: NameLike, value: Union[ValueLike, bool]

308 ) -> None:

309 """Set a configuration value.

310

311 Args:

312 section: Tuple with section name and optional subsection namee

313 name: Name of the configuration value, including section

314 and optional subsection

315 value: value of the setting

316 """

317 raise NotImplementedError(self.set)

318

319 def items(self, section: SectionLike) -> Iterator[tuple[Name, Value]]:

320 """Iterate over the configuration pairs for a specific section.

321

322 Args:

323 section: Tuple with section name and optional subsection namee

324 Returns:

325 Iterator over (name, value) pairs

326 """

327 raise NotImplementedError(self.items)

328

329 def sections(self) -> Iterator[Section]:

330 """Iterate over the sections.

331

332 Returns: Iterator over section tuples

333 """

334 raise NotImplementedError(self.sections)

335

336 def has_section(self, name: Section) -> bool:

337 """Check if a specified section exists.

338

339 Args:

340 name: Name of section to check for

341 Returns:

342 boolean indicating whether the section exists

343 """

344 return name in self.sections()

345

346

347class ConfigDict(Config, MutableMapping[Section, MutableMapping[Name, Value]]):

348 """Git configuration stored in a dictionary."""

349

350 def __init__(

351 self,

352 values: Union[

353 MutableMapping[Section, MutableMapping[Name, Value]], None

354 ] = None,

355 encoding: Union[str, None] = None,

356 ) -> None:

357 """Create a new ConfigDict."""

358 if encoding is None:

359 encoding = sys.getdefaultencoding()

360 self.encoding = encoding

361 self._values = CaseInsensitiveOrderedMultiDict.make(values)

362

363 def __repr__(self) -> str:

364 return f"{self.__class__.__name__}({self._values!r})"

365

366 def __eq__(self, other: object) -> bool:

367 return isinstance(other, self.__class__) and other._values == self._values

368

369 def __getitem__(self, key: Section) -> MutableMapping[Name, Value]:

370 return self._values.__getitem__(key)

371

372 def __setitem__(self, key: Section, value: MutableMapping[Name, Value]) -> None:

373 return self._values.__setitem__(key, value)

374

375 def __delitem__(self, key: Section) -> None:

376 return self._values.__delitem__(key)

377

378 def __iter__(self) -> Iterator[Section]:

379 return self._values.__iter__()

380

381 def __len__(self) -> int:

382 return self._values.__len__()

383

384 @classmethod

385 def _parse_setting(cls, name: str):

386 parts = name.split(".")

387 if len(parts) == 3:

388 return (parts[0], parts[1], parts[2])

389 else:

390 return (parts[0], None, parts[1])

391

392 def _check_section_and_name(

393 self, section: SectionLike, name: NameLike

394 ) -> tuple[Section, Name]:

395 if not isinstance(section, tuple):

396 section = (section,)

397

398 checked_section = tuple(

399 [

400 subsection.encode(self.encoding)

401 if not isinstance(subsection, bytes)

402 else subsection

403 for subsection in section

404 ]

405 )

406

407 if not isinstance(name, bytes):

408 name = name.encode(self.encoding)

409

410 return checked_section, name

411

412 def get_multivar(self, section: SectionLike, name: NameLike) -> Iterator[Value]:

413 section, name = self._check_section_and_name(section, name)

414

415 if len(section) > 1:

416 try:

417 return self._values[section].get_all(name)

418 except KeyError:

419 pass

420

421 return self._values[(section[0],)].get_all(name)

422

423 def get( # type: ignore[override]

424 self,

425 section: SectionLike,

426 name: NameLike,

427 ) -> Value:

428 section, name = self._check_section_and_name(section, name)

429

430 if len(section) > 1:

431 try:

432 return self._values[section][name]

433 except KeyError:

434 pass

435

436 return self._values[(section[0],)][name]

437

438 def set(

439 self,

440 section: SectionLike,

441 name: NameLike,

442 value: Union[ValueLike, bool],

443 ) -> None:

444 section, name = self._check_section_and_name(section, name)

445

446 if isinstance(value, bool):

447 value = b"true" if value else b"false"

448

449 if not isinstance(value, bytes):

450 value = value.encode(self.encoding)

451

452 section_dict = self._values.setdefault(section)

453 if hasattr(section_dict, "set"):

454 section_dict.set(name, value)

455 else:

456 section_dict[name] = value

457

458 def add(

459 self,

460 section: SectionLike,

461 name: NameLike,

462 value: Union[ValueLike, bool],

463 ) -> None:

464 """Add a value to a configuration setting, creating a multivar if needed."""

465 section, name = self._check_section_and_name(section, name)

466

467 if isinstance(value, bool):

468 value = b"true" if value else b"false"

469

470 if not isinstance(value, bytes):

471 value = value.encode(self.encoding)

472

473 self._values.setdefault(section)[name] = value

474

475 def items( # type: ignore[override]

476 self, section: Section

477 ) -> Iterator[tuple[Name, Value]]:

478 return self._values.get(section).items()

479

480 def sections(self) -> Iterator[Section]:

481 return self._values.keys()

482

483

484def _format_string(value: bytes) -> bytes:

485 if (

486 value.startswith((b" ", b"\t"))

487 or value.endswith((b" ", b"\t"))

488 or b"#" in value

489 ):

490 return b'"' + _escape_value(value) + b'"'

491 else:

492 return _escape_value(value)

493

494

495_ESCAPE_TABLE = {

496 ord(b"\\"): ord(b"\\"),

497 ord(b'"'): ord(b'"'),

498 ord(b"n"): ord(b"\n"),

499 ord(b"t"): ord(b"\t"),

500 ord(b"b"): ord(b"\b"),

501}

502_COMMENT_CHARS = [ord(b"#"), ord(b";")]

503_WHITESPACE_CHARS = [ord(b"\t"), ord(b" ")]

504

505

506def _parse_string(value: bytes) -> bytes:

507 value = bytearray(value.strip())

508 ret = bytearray()

509 whitespace = bytearray()

510 in_quotes = False

511 i = 0

512 while i < len(value):

513 c = value[i]

514 if c == ord(b"\\"):

515 i += 1

516 if i >= len(value):

517 # Backslash at end of string - treat as literal backslash

518 if whitespace:

519 ret.extend(whitespace)

520 whitespace = bytearray()

521 ret.append(ord(b"\\"))

522 else:

523 try:

524 v = _ESCAPE_TABLE[value[i]]

525 if whitespace:

526 ret.extend(whitespace)

527 whitespace = bytearray()

528 ret.append(v)

529 except KeyError:

530 # Unknown escape sequence - treat backslash as literal and process next char normally

531 if whitespace:

532 ret.extend(whitespace)

533 whitespace = bytearray()

534 ret.append(ord(b"\\"))

535 i -= 1 # Reprocess the character after the backslash

536 elif c == ord(b'"'):

537 in_quotes = not in_quotes

538 elif c in _COMMENT_CHARS and not in_quotes:

539 # the rest of the line is a comment

540 break

541 elif c in _WHITESPACE_CHARS:

542 whitespace.append(c)

543 else:

544 if whitespace:

545 ret.extend(whitespace)

546 whitespace = bytearray()

547 ret.append(c)

548 i += 1

549

550 if in_quotes:

551 raise ValueError("missing end quote")

552

553 return bytes(ret)

554

555

556def _escape_value(value: bytes) -> bytes:

557 """Escape a value."""

558 value = value.replace(b"\\", b"\\\\")

559 value = value.replace(b"\r", b"\\r")

560 value = value.replace(b"\n", b"\\n")

561 value = value.replace(b"\t", b"\\t")

562 value = value.replace(b'"', b'\\"')

563 return value

564

565

566def _check_variable_name(name: bytes) -> bool:

567 for i in range(len(name)):

568 c = name[i : i + 1]

569 if not c.isalnum() and c != b"-":

570 return False

571 return True

572

573

574def _check_section_name(name: bytes) -> bool:

575 for i in range(len(name)):

576 c = name[i : i + 1]

577 if not c.isalnum() and c not in (b"-", b"."):

578 return False

579 return True

580

581

582def _strip_comments(line: bytes) -> bytes:

583 comment_bytes = {ord(b"#"), ord(b";")}

584 quote = ord(b'"')

585 string_open = False

586 # Normalize line to bytearray for simple 2/3 compatibility

587 for i, character in enumerate(bytearray(line)):

588 # Comment characters outside balanced quotes denote comment start

589 if character == quote:

590 string_open = not string_open

591 elif not string_open and character in comment_bytes:

592 return line[:i]

593 return line

594

595

596def _is_line_continuation(value: bytes) -> bool:

597 """Check if a value ends with a line continuation backslash.

598

599 A line continuation occurs when a line ends with a backslash that is:

600 1. Not escaped (not preceded by another backslash)

601 2. Not within quotes

602

603 Args:

604 value: The value to check

605

606 Returns:

607 True if the value ends with a line continuation backslash

608 """

609 if not value.endswith((b"\\\n", b"\\\r\n")):

610 return False

611

612 # Remove only the newline characters, keep the content including the backslash

613 if value.endswith(b"\\\r\n"):

614 content = value[:-2] # Remove \r\n, keep the \

615 else:

616 content = value[:-1] # Remove \n, keep the \

617

618 if not content.endswith(b"\\"):

619 return False

620

621 # Count consecutive backslashes at the end

622 backslash_count = 0

623 for i in range(len(content) - 1, -1, -1):

624 if content[i : i + 1] == b"\\":

625 backslash_count += 1

626 else:

627 break

628

629 # If we have an odd number of backslashes, the last one is a line continuation

630 # If we have an even number, they are all escaped and there's no continuation

631 return backslash_count % 2 == 1

632

633

634def _parse_section_header_line(line: bytes) -> tuple[Section, bytes]:

635 # Parse section header ("[bla]")

636 line = _strip_comments(line).rstrip()

637 in_quotes = False

638 escaped = False

639 for i, c in enumerate(line):

640 if escaped:

641 escaped = False

642 continue

643 if c == ord(b'"'):

644 in_quotes = not in_quotes

645 if c == ord(b"\\"):

646 escaped = True

647 if c == ord(b"]") and not in_quotes:

648 last = i

649 break

650 else:

651 raise ValueError("expected trailing ]")

652 pts = line[1:last].split(b" ", 1)

653 line = line[last + 1 :]

654 section: Section

655 if len(pts) == 2:

656 # Handle subsections - Git allows more complex syntax for certain sections like includeIf

657 if pts[1][:1] == b'"' and pts[1][-1:] == b'"':

658 # Standard quoted subsection

659 pts[1] = pts[1][1:-1]

660 elif pts[0] == b"includeIf":

661 # Special handling for includeIf sections which can have complex conditions

662 # Git allows these without strict quote validation

663 pts[1] = pts[1].strip()

664 if pts[1][:1] == b'"' and pts[1][-1:] == b'"':

665 pts[1] = pts[1][1:-1]

666 else:

667 # Other sections must have quoted subsections

668 raise ValueError(f"Invalid subsection {pts[1]!r}")

669 if not _check_section_name(pts[0]):

670 raise ValueError(f"invalid section name {pts[0]!r}")

671 section = (pts[0], pts[1])

672 else:

673 if not _check_section_name(pts[0]):

674 raise ValueError(f"invalid section name {pts[0]!r}")

675 pts = pts[0].split(b".", 1)

676 if len(pts) == 2:

677 section = (pts[0], pts[1])

678 else:

679 section = (pts[0],)

680 return section, line

681

682

683class ConfigFile(ConfigDict):

684 """A Git configuration file, like .git/config or ~/.gitconfig."""

685

686 def __init__(

687 self,

688 values: Union[

689 MutableMapping[Section, MutableMapping[Name, Value]], None

690 ] = None,

691 encoding: Union[str, None] = None,

692 ) -> None:

693 super().__init__(values=values, encoding=encoding)

694 self.path: Optional[str] = None

695 self._included_paths: set[str] = set() # Track included files to prevent cycles

696

697 @classmethod

698 def from_file(

699 cls,

700 f: BinaryIO,

701 *,

702 config_dir: Optional[str] = None,

703 included_paths: Optional[set[str]] = None,

704 include_depth: int = 0,

705 max_include_depth: int = DEFAULT_MAX_INCLUDE_DEPTH,

706 file_opener: Optional[FileOpener] = None,

707 condition_matchers: Optional[dict[str, ConditionMatcher]] = None,

708 ) -> "ConfigFile":

709 """Read configuration from a file-like object.

710

711 Args:

712 f: File-like object to read from

713 config_dir: Directory containing the config file (for relative includes)

714 included_paths: Set of already included paths (to prevent cycles)

715 include_depth: Current include depth (to prevent infinite recursion)

716 max_include_depth: Maximum allowed include depth

717 file_opener: Optional callback to open included files

718 condition_matchers: Optional dict of condition matchers for includeIf

719 """

720 if include_depth > max_include_depth:

721 # Prevent excessive recursion

722 raise ValueError(f"Maximum include depth ({max_include_depth}) exceeded")

723

724 ret = cls()

725 if included_paths is not None:

726 ret._included_paths = included_paths.copy()

727

728 section: Optional[Section] = None

729 setting = None

730 continuation = None

731 for lineno, line in enumerate(f.readlines()):

732 if lineno == 0 and line.startswith(b"\xef\xbb\xbf"):

733 line = line[3:]

734 line = line.lstrip()

735 if setting is None:

736 if len(line) > 0 and line[:1] == b"[":

737 section, line = _parse_section_header_line(line)

738 ret._values.setdefault(section)

739 if _strip_comments(line).strip() == b"":

740 continue

741 if section is None:

742 raise ValueError(f"setting {line!r} without section")

743 try:

744 setting, value = line.split(b"=", 1)

745 except ValueError:

746 setting = line

747 value = b"true"

748 setting = setting.strip()

749 if not _check_variable_name(setting):

750 raise ValueError(f"invalid variable name {setting!r}")

751 if _is_line_continuation(value):

752 if value.endswith(b"\\\r\n"):

753 continuation = value[:-3]

754 else:

755 continuation = value[:-2]

756 else:

757 continuation = None

758 value = _parse_string(value)

759 ret._values[section][setting] = value

760

761 # Process include/includeIf directives

762 ret._handle_include_directive(

763 section,

764 setting,

765 value,

766 config_dir=config_dir,

767 include_depth=include_depth,

768 max_include_depth=max_include_depth,

769 file_opener=file_opener,

770 condition_matchers=condition_matchers,

771 )

772

773 setting = None

774 else: # continuation line

775 assert continuation is not None

776 if _is_line_continuation(line):

777 if line.endswith(b"\\\r\n"):

778 continuation += line[:-3]

779 else:

780 continuation += line[:-2]

781 else:

782 continuation += line

783 value = _parse_string(continuation)

784 ret._values[section][setting] = value

785

786 # Process include/includeIf directives

787 ret._handle_include_directive(

788 section,

789 setting,

790 value,

791 config_dir=config_dir,

792 include_depth=include_depth,

793 max_include_depth=max_include_depth,

794 file_opener=file_opener,

795 condition_matchers=condition_matchers,

796 )

797

798 continuation = None

799 setting = None

800 return ret

801

802 def _handle_include_directive(

803 self,

804 section: Optional[Section],

805 setting: bytes,

806 value: bytes,

807 *,

808 config_dir: Optional[str],

809 include_depth: int,

810 max_include_depth: int,

811 file_opener: Optional[FileOpener],

812 condition_matchers: Optional[dict[str, ConditionMatcher]],

813 ) -> None:

814 """Handle include/includeIf directives during config parsing."""

815 if (

816 section is not None

817 and setting == b"path"

818 and (

819 section[0].lower() == b"include"

820 or (len(section) > 1 and section[0].lower() == b"includeif")

821 )

822 ):

823 self._process_include(

824 section,

825 value,

826 config_dir=config_dir,

827 include_depth=include_depth,

828 max_include_depth=max_include_depth,

829 file_opener=file_opener,

830 condition_matchers=condition_matchers,

831 )

832

833 def _process_include(

834 self,

835 section: Section,

836 path_value: bytes,

837 *,

838 config_dir: Optional[str],

839 include_depth: int,

840 max_include_depth: int,

841 file_opener: Optional[FileOpener],

842 condition_matchers: Optional[dict[str, ConditionMatcher]],

843 ) -> None:

844 """Process an include or includeIf directive."""

845 path_str = path_value.decode(self.encoding, errors="replace")

846

847 # Handle includeIf conditions

848 if len(section) > 1 and section[0].lower() == b"includeif":

849 condition = section[1].decode(self.encoding, errors="replace")

850 if not self._evaluate_includeif_condition(

851 condition, config_dir, condition_matchers

852 ):

853 return

854

855 # Resolve the include path

856 include_path = self._resolve_include_path(path_str, config_dir)

857 if not include_path:

858 return

859

860 # Check for circular includes

861 try:

862 abs_path = str(Path(include_path).resolve())

863 except (OSError, ValueError) as e:

864 # Invalid path - log and skip

865 logger.debug("Invalid include path %r: %s", include_path, e)

866 return

867 if abs_path in self._included_paths:

868 return

869

870 # Load and merge the included file

871 try:

872 # Use provided file opener or default to GitFile

873 if file_opener is None:

874

875 def opener(path):

876 return GitFile(path, "rb")

877 else:

878 opener = file_opener

879

880 f = opener(include_path)

881 except (OSError, ValueError) as e:

882 # Git silently ignores missing or unreadable include files

883 # Log for debugging purposes

884 logger.debug("Invalid include path %r: %s", include_path, e)

885 else:

886 with f as included_file:

887 # Track this path to prevent cycles

888 self._included_paths.add(abs_path)

889

890 # Parse the included file

891 included_config = ConfigFile.from_file(

892 included_file,

893 config_dir=os.path.dirname(include_path),

894 included_paths=self._included_paths,

895 include_depth=include_depth + 1,

896 max_include_depth=max_include_depth,

897 file_opener=file_opener,

898 condition_matchers=condition_matchers,

899 )

900

901 # Merge the included configuration

902 self._merge_config(included_config)

903

904 def _merge_config(self, other: "ConfigFile") -> None:

905 """Merge another config file into this one."""

906 for section, values in other._values.items():

907 if section not in self._values:

908 self._values[section] = CaseInsensitiveOrderedMultiDict()

909 for key, value in values.items():

910 self._values[section][key] = value

911

912 def _resolve_include_path(

913 self, path: str, config_dir: Optional[str]

914 ) -> Optional[str]:

915 """Resolve an include path to an absolute path."""

916 # Expand ~ to home directory

917 path = os.path.expanduser(path)

918

919 # If path is relative and we have a config directory, make it relative to that

920 if not os.path.isabs(path) and config_dir:

921 path = os.path.join(config_dir, path)

922

923 return path

924

925 def _evaluate_includeif_condition(

926 self,

927 condition: str,

928 config_dir: Optional[str] = None,

929 condition_matchers: Optional[dict[str, ConditionMatcher]] = None,

930 ) -> bool:

931 """Evaluate an includeIf condition."""

932 # Try custom matchers first if provided

933 if condition_matchers:

934 for prefix, matcher in condition_matchers.items():

935 if condition.startswith(prefix):

936 return matcher(condition[len(prefix) :])

937

938 # Fall back to built-in matchers

939 if condition.startswith("hasconfig:"):

940 return self._evaluate_hasconfig_condition(condition[10:])

941 else:

942 # Unknown condition type - log and ignore (Git behavior)

943 logger.debug("Unknown includeIf condition: %r", condition)

944 return False

945

946 def _evaluate_hasconfig_condition(self, condition: str) -> bool:

947 """Evaluate a hasconfig condition.

948

949 Format: hasconfig:config.key:pattern

950 Example: hasconfig:remote.*.url:ssh://org-*@github.com/**

951 """

952 # Split on the first colon to separate config key from pattern

953 parts = condition.split(":", 1)

954 if len(parts) != 2:

955 logger.debug("Invalid hasconfig condition format: %r", condition)

956 return False

957

958 config_key, pattern = parts

959

960 # Parse the config key to get section and name

961 key_parts = config_key.split(".", 2)

962 if len(key_parts) < 2:

963 logger.debug("Invalid hasconfig config key: %r", config_key)

964 return False

965

966 # Handle wildcards in section names (e.g., remote.*)

967 if len(key_parts) == 3 and key_parts[1] == "*":

968 # Match any subsection

969 section_prefix = key_parts[0].encode(self.encoding)

970 name = key_parts[2].encode(self.encoding)

971

972 # Check all sections that match the pattern

973 for section in self.sections():

974 if len(section) == 2 and section[0] == section_prefix:

975 try:

976 values = list(self.get_multivar(section, name))

977 for value in values:

978 if self._match_hasconfig_pattern(value, pattern):

979 return True

980 except KeyError:

981 continue

982 else:

983 # Direct section lookup

984 if len(key_parts) == 2:

985 section = (key_parts[0].encode(self.encoding),)

986 name = key_parts[1].encode(self.encoding)

987 else:

988 section = (

989 key_parts[0].encode(self.encoding),

990 key_parts[1].encode(self.encoding),

991 )

992 name = key_parts[2].encode(self.encoding)

993

994 try:

995 values = list(self.get_multivar(section, name))

996 for value in values:

997 if self._match_hasconfig_pattern(value, pattern):

998 return True

999 except KeyError:

1000 pass

1001

1002 return False

1003

1004 def _match_hasconfig_pattern(self, value: bytes, pattern: str) -> bool:

1005 """Match a config value against a hasconfig pattern.

1006

1007 Supports simple glob patterns like ``*`` and ``**``.

1008 """

1009 value_str = value.decode(self.encoding, errors="replace")

1010 return match_glob_pattern(value_str, pattern)

1011

1012 @classmethod

1013 def from_path(

1014 cls,

1015 path: Union[str, os.PathLike],

1016 *,

1017 max_include_depth: int = DEFAULT_MAX_INCLUDE_DEPTH,

1018 file_opener: Optional[FileOpener] = None,

1019 condition_matchers: Optional[dict[str, ConditionMatcher]] = None,

1020 ) -> "ConfigFile":

1021 """Read configuration from a file on disk.

1022

1023 Args:

1024 path: Path to the configuration file

1025 max_include_depth: Maximum allowed include depth

1026 file_opener: Optional callback to open included files

1027 condition_matchers: Optional dict of condition matchers for includeIf

1028 """

1029 abs_path = os.fspath(path)

1030 config_dir = os.path.dirname(abs_path)

1031

1032 # Use provided file opener or default to GitFile

1033 if file_opener is None:

1034

1035 def opener(p):

1036 return GitFile(p, "rb")

1037 else:

1038 opener = file_opener

1039

1040 with opener(abs_path) as f:

1041 ret = cls.from_file(

1042 f,

1043 config_dir=config_dir,

1044 max_include_depth=max_include_depth,

1045 file_opener=file_opener,

1046 condition_matchers=condition_matchers,

1047 )

1048 ret.path = abs_path

1049 return ret

1050

1051 def write_to_path(self, path: Optional[Union[str, os.PathLike]] = None) -> None:

1052 """Write configuration to a file on disk."""

1053 if path is None:

1054 if self.path is None:

1055 raise ValueError("No path specified and no default path available")

1056 path_to_use: Union[str, os.PathLike] = self.path

1057 else:

1058 path_to_use = path

1059 with GitFile(path_to_use, "wb") as f:

1060 self.write_to_file(f)

1061

1062 def write_to_file(self, f: BinaryIO) -> None:

1063 """Write configuration to a file-like object."""

1064 for section, values in self._values.items():

1065 try:

1066 section_name, subsection_name = section

1067 except ValueError:

1068 (section_name,) = section

1069 subsection_name = None

1070 if subsection_name is None:

1071 f.write(b"[" + section_name + b"]\n")

1072 else:

1073 f.write(b"[" + section_name + b' "' + subsection_name + b'"]\n')

1074 for key, value in values.items():

1075 value = _format_string(value)

1076 f.write(b"\t" + key + b" = " + value + b"\n")

1077

1078

1079def get_xdg_config_home_path(*path_segments):

1080 xdg_config_home = os.environ.get(

1081 "XDG_CONFIG_HOME",

1082 os.path.expanduser("~/.config/"),

1083 )

1084 return os.path.join(xdg_config_home, *path_segments)

1085

1086

1087def _find_git_in_win_path():

1088 for exe in ("git.exe", "git.cmd"):

1089 for path in os.environ.get("PATH", "").split(";"):

1090 if os.path.exists(os.path.join(path, exe)):

1091 # in windows native shells (powershell/cmd) exe path is

1092 # .../Git/bin/git.exe or .../Git/cmd/git.exe

1093 #

1094 # in git-bash exe path is .../Git/mingw64/bin/git.exe

1095 git_dir, _bin_dir = os.path.split(path)

1096 yield git_dir

1097 parent_dir, basename = os.path.split(git_dir)

1098 if basename == "mingw32" or basename == "mingw64":

1099 yield parent_dir

1100 break

1101

1102

1103def _find_git_in_win_reg():

1104 import platform

1105 import winreg

1106

1107 if platform.machine() == "AMD64":

1108 subkey = (

1109 "SOFTWARE\\Wow6432Node\\Microsoft\\Windows\\"

1110 "CurrentVersion\\Uninstall\\Git_is1"

1111 )

1112 else:

1113 subkey = "SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\Uninstall\\Git_is1"

1114

1115 for key in (winreg.HKEY_CURRENT_USER, winreg.HKEY_LOCAL_MACHINE): # type: ignore

1116 with suppress(OSError):

1117 with winreg.OpenKey(key, subkey) as k: # type: ignore

1118 val, typ = winreg.QueryValueEx(k, "InstallLocation") # type: ignore

1119 if typ == winreg.REG_SZ: # type: ignore

1120 yield val

1121

1122

1123# There is no set standard for system config dirs on windows. We try the

1124# following:

1125# - %PROGRAMDATA%/Git/config - (deprecated) Windows config dir per CGit docs

1126# - %PROGRAMFILES%/Git/etc/gitconfig - Git for Windows (msysgit) config dir

1127# Used if CGit installation (Git/bin/git.exe) is found in PATH in the

1128# system registry

1129def get_win_system_paths():

1130 if "PROGRAMDATA" in os.environ:

1131 yield os.path.join(os.environ["PROGRAMDATA"], "Git", "config")

1132

1133 for git_dir in _find_git_in_win_path():

1134 yield os.path.join(git_dir, "etc", "gitconfig")

1135 for git_dir in _find_git_in_win_reg():

1136 yield os.path.join(git_dir, "etc", "gitconfig")

1137

1138

1139class StackedConfig(Config):

1140 """Configuration which reads from multiple config files.."""

1141

1142 def __init__(

1143 self, backends: list[ConfigFile], writable: Optional[ConfigFile] = None

1144 ) -> None:

1145 self.backends = backends

1146 self.writable = writable

1147

1148 def __repr__(self) -> str:

1149 return f"<{self.__class__.__name__} for {self.backends!r}>"

1150

1151 @classmethod

1152 def default(cls) -> "StackedConfig":

1153 return cls(cls.default_backends())

1154

1155 @classmethod

1156 def default_backends(cls) -> list[ConfigFile]:

1157 """Retrieve the default configuration.

1158

1159 See git-config(1) for details on the files searched.

1160 """

1161 paths = []

1162 paths.append(os.path.expanduser("~/.gitconfig"))

1163 paths.append(get_xdg_config_home_path("git", "config"))

1164

1165 if "GIT_CONFIG_NOSYSTEM" not in os.environ:

1166 paths.append("/etc/gitconfig")

1167 if sys.platform == "win32":

1168 paths.extend(get_win_system_paths())

1169

1170 backends = []

1171 for path in paths:

1172 try:

1173 cf = ConfigFile.from_path(path)

1174 except FileNotFoundError:

1175 continue

1176 backends.append(cf)

1177 return backends

1178

1179 def get(self, section: SectionLike, name: NameLike) -> Value:

1180 if not isinstance(section, tuple):

1181 section = (section,)

1182 for backend in self.backends:

1183 try:

1184 return backend.get(section, name)

1185 except KeyError:

1186 pass

1187 raise KeyError(name)

1188

1189 def get_multivar(self, section: SectionLike, name: NameLike) -> Iterator[Value]:

1190 if not isinstance(section, tuple):

1191 section = (section,)

1192 for backend in self.backends:

1193 try:

1194 yield from backend.get_multivar(section, name)

1195 except KeyError:

1196 pass

1197

1198 def set(

1199 self, section: SectionLike, name: NameLike, value: Union[ValueLike, bool]

1200 ) -> None:

1201 if self.writable is None:

1202 raise NotImplementedError(self.set)

1203 return self.writable.set(section, name, value)

1204

1205 def sections(self) -> Iterator[Section]:

1206 seen = set()

1207 for backend in self.backends:

1208 for section in backend.sections():

1209 if section not in seen:

1210 seen.add(section)

1211 yield section

1212

1213

1214def read_submodules(

1215 path: Union[str, os.PathLike],

1216) -> Iterator[tuple[bytes, bytes, bytes]]:

1217 """Read a .gitmodules file."""

1218 cfg = ConfigFile.from_path(path)

1219 return parse_submodules(cfg)

1220

1221

1222def parse_submodules(config: ConfigFile) -> Iterator[tuple[bytes, bytes, bytes]]:

1223 """Parse a gitmodules GitConfig file, returning submodules.

1224

1225 Args:

1226 config: A `ConfigFile`

1227 Returns:

1228 list of tuples (submodule path, url, name),

1229 where name is quoted part of the section's name.

1230 """

1231 for section in config.keys():

1232 section_kind, section_name = section

1233 if section_kind == b"submodule":

1234 try:

1235 sm_path = config.get(section, b"path")

1236 sm_url = config.get(section, b"url")

1237 yield (sm_path, sm_url, section_name)

1238 except KeyError:

1239 # If either path or url is missing, just ignore this

1240 # submodule entry and move on to the next one. This is

1241 # how git itself handles malformed .gitmodule entries.

1242 pass

1243

1244

1245def iter_instead_of(config: Config, push: bool = False) -> Iterable[tuple[str, str]]:

1246 """Iterate over insteadOf / pushInsteadOf values."""

1247 for section in config.sections():

1248 if section[0] != b"url":

1249 continue

1250 replacement = section[1]

1251 try:

1252 needles = list(config.get_multivar(section, "insteadOf"))

1253 except KeyError:

1254 needles = []

1255 if push:

1256 try:

1257 needles += list(config.get_multivar(section, "pushInsteadOf"))

1258 except KeyError:

1259 pass

1260 for needle in needles:

1261 assert isinstance(needle, bytes)

1262 yield needle.decode("utf-8"), replacement.decode("utf-8")

1263

1264

1265def apply_instead_of(config: Config, orig_url: str, push: bool = False) -> str:

1266 """Apply insteadOf / pushInsteadOf to a URL."""

1267 longest_needle = ""

1268 updated_url = orig_url

1269 for needle, replacement in iter_instead_of(config, push):

1270 if not orig_url.startswith(needle):

1271 continue

1272 if len(longest_needle) < len(needle):

1273 longest_needle = needle

1274 updated_url = replacement + orig_url[len(needle) :]

1275 return updated_url