Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion src/scmrepo/git/backend/dulwich/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -68,7 +68,8 @@ def _get_auth(self) -> Dict[str, str]:
from urllib3.util import make_headers

try:
creds = Credential(username=self._username, url=self._base_url).fill()
base_url = self._base_url.rstrip("/")
creds = Credential(username=self._username, url=base_url).fill()
self._store_credentials = creds
return make_headers(basic_auth=f"{creds.username}:{creds.password}")
except CredentialNotFoundError:
Expand Down
45 changes: 34 additions & 11 deletions src/scmrepo/git/credentials.py
Original file line number Diff line number Diff line change
Expand Up @@ -107,14 +107,15 @@ class GitCredentialHelper(CredentialHelper):
>>> password = credentials.password
"""

def __init__(self, command: str):
def __init__(self, command: str, use_http_path: bool = False):
super().__init__()
self._command = command
self._run_kwargs: Dict[str, Any] = {}
if self._command[0] == "!":
# On Windows this will only work in git-bash and/or WSL2
self._run_kwargs["shell"] = True
self._encoding = locale.getpreferredencoding()
self.use_http_path = use_http_path

def _prepare_command(self, action: Optional[str] = None) -> Union[str, List[str]]:
if self._command[0] == "!":
Expand Down Expand Up @@ -150,7 +151,12 @@ def get(self, credential: "Credential", **kwargs) -> "Credential":
if not (credential.protocol or credential.host):
raise ValueError("One of protocol, hostname must be provided")
cmd = self._prepare_command("get")
helper_input = [f"{key}={value}" for key, value in credential.items()]
use_path = credential.protocol in ("http", "https") and self.use_http_path
helper_input = [
f"{key}={value}"
for key, value in credential.items()
if key != "path" or use_path
]
helper_input.append("")

try:
Expand All @@ -170,9 +176,9 @@ def get(self, credential: "Credential", **kwargs) -> "Credential":
logger.debug(res.stderr)

credentials = {}
for line in res.stdout.strip().splitlines():
for line in res.stdout.splitlines():
try:
key, value = line.split("=")
key, value = line.split("=", maxsplit=1)
credentials[key] = value
except ValueError:
continue
Expand All @@ -186,7 +192,12 @@ def get(self, credential: "Credential", **kwargs) -> "Credential":
def store(self, credential: "Credential", **kwargs):
"""Store the credential, if applicable to the helper"""
cmd = self._prepare_command("store")
helper_input = [f"{key}={value}" for key, value in credential.items()]
use_path = credential.protocol in ("http", "https") and self.use_http_path
helper_input = [
f"{key}={value}"
for key, value in credential.items()
if key != "path" or use_path
]
helper_input.append("")

try:
Expand All @@ -205,7 +216,12 @@ def store(self, credential: "Credential", **kwargs):
def erase(self, credential: "Credential", **kwargs):
"""Remove a matching credential, if any, from the helper’s storage"""
cmd = self._prepare_command("erase")
helper_input = [f"{key}={value}" for key, value in credential.items()]
use_path = credential.protocol in ("http", "https") and self.use_http_path
helper_input = [
f"{key}={value}"
for key, value in credential.items()
if key != "path" or use_path
]
helper_input.append("")

try:
Expand All @@ -224,7 +240,7 @@ def erase(self, credential: "Credential", **kwargs):
@staticmethod
def get_matching_commands(
base_url: str, config: Optional[Union["ConfigDict", "StackedConfig"]] = None
):
) -> Iterator[Tuple[str, bool]]:
config = config or StackedConfig.default()
if isinstance(config, StackedConfig):
backends: Iterable["ConfigDict"] = config.backends
Expand All @@ -240,7 +256,11 @@ def get_matching_commands(
except KeyError:
# no helper configured
continue
yield command.decode(conf.encoding or sys.getdefaultencoding())
use_http_path = conf.get_boolean(section, "usehttppath", False)
yield (
command.decode(conf.encoding or sys.getdefaultencoding()),
use_http_path,
)


class _CredentialKey(NamedTuple):
Expand Down Expand Up @@ -379,7 +399,8 @@ def _get_interactive(

def store(self, credential: "Credential", **kwargs):
"""Store the credential, if applicable to the helper"""
self[credential] = credential
if credential.protocol or credential.host or credential.path:
self[credential] = credential

def erase(self, credential: "Credential", **kwargs):
"""Remove a matching credential, if any, from the helper’s storage"""
Expand Down Expand Up @@ -493,6 +514,8 @@ def __init__(
self.host = self.host or f"{hostname}{port}"
self.username = self.username or parsed.username
self.password = self.password or parsed.password
if parsed.path:
self.path = self.path or parsed.path.lstrip("/")

def __getitem__(self, key: object) -> str:
if isinstance(key, str):
Expand Down Expand Up @@ -540,8 +563,8 @@ def describe(self, use_path: bool = False, sanitize: bool = True) -> str:
def helpers(self) -> List["CredentialHelper"]:
url = self.url
return [
GitCredentialHelper(command)
for command in GitCredentialHelper.get_matching_commands(url)
GitCredentialHelper(command, use_http_path=use_http_path)
for command, use_http_path in GitCredentialHelper.get_matching_commands(url)
]

def fill(self, interactive: bool = True) -> "Credential":
Expand Down
52 changes: 51 additions & 1 deletion tests/test_credentials.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
import io
import os

import pytest
Expand Down Expand Up @@ -25,14 +26,30 @@ def test_subprocess_get(git_helper, mocker):
)
),
)
creds = git_helper.get(Credential(protocol="https", host="foo.com"))
creds = git_helper.get(Credential(protocol="https", host="foo.com", path="foo.git"))
assert run.call_args.args[0] == ["git-credential-foo", "get"]
assert run.call_args.kwargs.get("input") == os.linesep.join(
["protocol=https", "host=foo.com", ""]
)
assert creds == Credential(url="https://foo:bar@foo.com")


def test_subprocess_get_use_http_path(git_helper, mocker):
git_helper.use_http_path = True
run = mocker.patch(
"subprocess.run",
return_value=mocker.Mock(
stdout=os.linesep.join(["username=foo", "password=bar", ""])
),
)
creds = git_helper.get(Credential(protocol="https", host="foo.com", path="foo.git"))
assert run.call_args.args[0] == ["git-credential-foo", "get"]
assert run.call_args.kwargs.get("input") == os.linesep.join(
["protocol=https", "host=foo.com", "path=foo.git", ""]
)
assert creds == Credential(username="foo", password="bar")


def test_subprocess_get_failed(git_helper, mocker):
from subprocess import CalledProcessError

Expand Down Expand Up @@ -151,3 +168,36 @@ def test_memory_helper_prompt_askpass(mocker):
"/usr/local/bin/my-askpass",
"Password for 'https://foo@foo.com': ",
]


def test_get_matching_commands():
from dulwich.config import ConfigFile

config_file = io.BytesIO(
"""
[credential]
helper = /usr/local/bin/my-helper
UseHttpPath = true
""".encode(
"ascii"
)
)
config_file.seek(0)
config = ConfigFile.from_file(config_file)
assert list(
GitCredentialHelper.get_matching_commands("https://foo.com/foo.git", config)
) == [("/usr/local/bin/my-helper", True)]

config_file = io.BytesIO(
"""
[credential]
helper = /usr/local/bin/my-helper
""".encode(
"ascii"
)
)
config_file.seek(0)
config = ConfigFile.from_file(config_file)
assert list(
GitCredentialHelper.get_matching_commands("https://foo.com/foo.git", config)
) == [("/usr/local/bin/my-helper", False)]