|
6 | 6 | import os |
7 | 7 | import shutil |
8 | 8 | import subprocess |
9 | | -from typing import TYPE_CHECKING, Optional, Tuple, cast |
| 9 | +from typing import TYPE_CHECKING, Literal, Optional, Tuple, cast |
10 | 10 | from urllib.parse import SplitResult, urlparse, urlsplit |
11 | 11 |
|
12 | 12 | from httpx import URL, Auth, BasicAuth |
@@ -83,20 +83,31 @@ def __init__(self, cmd: str) -> None: |
83 | 83 | self.keyring = cmd |
84 | 84 |
|
85 | 85 | def get_auth_info(self, url: str, username: str | None) -> AuthInfo | None: |
| 86 | + logger.debug("Getting credentials from keyring CLI for url: %s", url) |
| 87 | + cred = self._get_secret(url, username or "", mode="creds") |
| 88 | + if cred is not None: |
| 89 | + username, password = cred.splitlines() |
| 90 | + return username, password |
| 91 | + |
86 | 92 | if username is None: |
87 | 93 | username = "__token__" |
88 | 94 | logger.debug("Getting password from keyring CLI for %s@%s", username, url) |
89 | | - password = self._get_password(url, username) |
| 95 | + password = self._get_secret(url, username) # type: ignore[assignment] |
90 | 96 | if password is not None: |
91 | 97 | return username, password |
92 | 98 | return None |
93 | 99 |
|
94 | 100 | def save_auth_info(self, url: str, username: str, password: str) -> None: |
95 | 101 | return self._set_password(url, username, password) |
96 | 102 |
|
97 | | - def _get_password(self, service_name: str, username: str) -> str | None: |
98 | | - """Mirror the implementation of keyring.get_password using cli""" |
99 | | - cmd = [self.keyring, "get", service_name, username] |
| 103 | + def _get_secret( |
| 104 | + self, |
| 105 | + service_name: str, |
| 106 | + username: str, |
| 107 | + mode: Literal["password", "creds"] = "password", |
| 108 | + ) -> str | None: |
| 109 | + """Mirror the implementation of keyring.get_[password|credential] using cli""" |
| 110 | + cmd = [self.keyring, f"--mode={mode}", "get", service_name, username] |
100 | 111 | env = dict(os.environ, PYTHONIOENCODING="utf-8") |
101 | 112 | res = subprocess.run( |
102 | 113 | cmd, stdin=subprocess.DEVNULL, capture_output=True, env=env |
|
0 commit comments