Skip to content

Commit

Permalink
refactor PoetryKeyring
Browse files Browse the repository at this point in the history
  • Loading branch information
Popkornium18 committed Feb 3, 2024
1 parent 981d411 commit 753ab64
Show file tree
Hide file tree
Showing 3 changed files with 105 additions and 102 deletions.
14 changes: 6 additions & 8 deletions src/poetry/utils/authenticator.py
Original file line number Diff line number Diff line change
Expand Up @@ -90,17 +90,15 @@ def get_http_credentials(
**(password_manager.get_http_auth(self.name) or {})
)

if credential.password is None:
if credential.password is not None:
return credential

if password_manager.use_keyring:
# fallback to url and netloc based keyring entries
credential = password_manager.keyring.get_credential(
credential = password_manager.get_credential(
self.url, self.netloc, username=credential.username
)

if credential.password is not None:
return HTTPAuthCredential(
username=credential.username, password=credential.password
)

return credential


Expand Down Expand Up @@ -305,7 +303,7 @@ def _get_credentials_for_url(
if credential.password is None:
parsed_url = urllib.parse.urlsplit(url)
netloc = parsed_url.netloc
credential = self._password_manager.keyring.get_credential(
credential = self._password_manager.get_credential(
url, netloc, username=credential.username
)

Expand Down
159 changes: 82 additions & 77 deletions src/poetry/utils/password_manager.py
Original file line number Diff line number Diff line change
@@ -1,13 +1,16 @@
from __future__ import annotations

import dataclasses
import functools
import logging

from contextlib import suppress
from typing import TYPE_CHECKING


if TYPE_CHECKING:
from keyring.backend import KeyringBackend

from poetry.config.config import Config

logger = logging.getLogger(__name__)
Expand All @@ -30,21 +33,10 @@ class HTTPAuthCredential:
class PoetryKeyring:
def __init__(self, namespace: str) -> None:
self._namespace = namespace
self._is_available = True

self._check()

def is_available(self) -> bool:
return self._is_available

def get_credential(
self, *names: str, username: str | None = None
) -> HTTPAuthCredential:
default = HTTPAuthCredential(username=username, password=None)

if not self.is_available():
return default

import keyring

from keyring.errors import KeyringError
Expand All @@ -64,12 +56,9 @@ def get_credential(
username=credential.username, password=credential.password
)

return default
return HTTPAuthCredential(username=username, password=None)

def get_password(self, name: str, username: str) -> str | None:
if not self.is_available():
return None

import keyring
import keyring.errors

Expand All @@ -83,9 +72,6 @@ def get_password(self, name: str, username: str) -> str | None:
)

def set_password(self, name: str, username: str, password: str) -> None:
if not self.is_available():
return

import keyring
import keyring.errors

Expand All @@ -99,9 +85,6 @@ def set_password(self, name: str, username: str, password: str) -> None:
)

def delete_password(self, name: str, username: str) -> None:
if not self.is_available():
return

import keyring.errors

name = self.get_entry_name(name)
Expand All @@ -116,69 +99,75 @@ def delete_password(self, name: str, username: str) -> None:
def get_entry_name(self, name: str) -> str:
return f"{self._namespace}-{name}"

def _check(self) -> None:
@classmethod
def is_available(cls) -> bool:
logger.debug("Checking if keyring is available")
try:
import keyring
import keyring.backend
except ImportError as e:
logger.debug("An error occurred while importing keyring: %s", e)
self._is_available = False
return False

return
def backend_name(backend: KeyringBackend) -> str:
name: str = backend.name
return name.split(" ")[0]

backend = keyring.get_keyring()
name = backend.name.split(" ")[0]
if name in ("fail", "null"):
logger.debug("No suitable keyring backend found")
self._is_available = False
elif "plaintext" in backend.name.lower():
logger.debug("Only a plaintext keyring backend is available. Not using it.")
self._is_available = False
elif name == "chainer":
try:
import keyring.backend
def backend_is_valid(backend: KeyringBackend) -> bool:
name = backend_name(backend)
if name in ("chainer", "fail", "null"):
logger.debug(f"Backend {backend.name!r} is not suitable")
return False
elif "plaintext" in backend.name.lower():
logger.debug(f"Not using plaintext keyring backend {backend.name!r}")
return False

backends = keyring.backend.get_all_keyring()
return True

self._is_available = any(
b.name.split(" ")[0] not in ["chainer", "fail", "null"]
and "plaintext" not in b.name.lower()
for b in backends
)
except ImportError:
self._is_available = False
backend = keyring.get_keyring()
if backend_name(backend) == "chainer":
backends = keyring.backend.get_all_keyring()
valid_backend = next((b for b in backends if backend_is_valid(b)), None)
else:
valid_backend = backend if backend_is_valid(backend) else None

if not self._is_available:
logger.debug("No suitable keyring backends were found")
if valid_backend is None:
logger.debug("No valid keyring backend was found")
return False
else:
logger.debug(f"Using keyring backend {backend.name!r}")
return True


class PasswordManager:
def __init__(self, config: Config) -> None:
self._config = config
self._keyring: PoetryKeyring | None = None

@property
def keyring(self) -> PoetryKeyring:
if self._keyring is None:
self._keyring = PoetryKeyring("poetry-repository")
@functools.cached_property
def use_keyring(self) -> bool:
return PoetryKeyring.is_available()

if not self._keyring.is_available():
logger.debug(
"<warning>Keyring is not available, credentials will be stored and "
"retrieved from configuration files as plaintext.</>"
)
@functools.cached_property
def keyring(self) -> PoetryKeyring:
if not self.use_keyring:
raise PoetryKeyringError(
"Access to keyring was requested, but it is not available"
)

return self._keyring
return PoetryKeyring("poetry-repository")

@staticmethod
def warn_plaintext_credentials_stored() -> None:
logger.warning("Using a plaintext file to store credentials")

def set_pypi_token(self, name: str, token: str) -> None:
if not self.keyring.is_available():
def set_pypi_token(self, repo_name: str, token: str) -> None:
if not self.use_keyring:
self.warn_plaintext_credentials_stored()
self._config.auth_config_source.add_property(f"pypi-token.{name}", token)
self._config.auth_config_source.add_property(
f"pypi-token.{repo_name}", token
)
else:
self.keyring.set_password(name, "__token__", token)
self.keyring.set_password(repo_name, "__token__", token)

def get_pypi_token(self, repo_name: str) -> str | None:
"""Get PyPi token.
Expand All @@ -194,41 +183,49 @@ def get_pypi_token(self, repo_name: str) -> str | None:
if token:
return token

return self.keyring.get_password(repo_name, "__token__")
if self.use_keyring:
return self.keyring.get_password(repo_name, "__token__")
else:
return None

def delete_pypi_token(self, name: str) -> None:
if not self.keyring.is_available():
return self._config.auth_config_source.remove_property(f"pypi-token.{name}")
def delete_pypi_token(self, repo_name: str) -> None:
if not self.use_keyring:
return self._config.auth_config_source.remove_property(
f"pypi-token.{repo_name}"
)

self.keyring.delete_password(name, "__token__")
self.keyring.delete_password(repo_name, "__token__")

def get_http_auth(self, name: str) -> dict[str, str | None] | None:
username = self._config.get(f"http-basic.{name}.username")
password = self._config.get(f"http-basic.{name}.password")
def get_http_auth(self, repo_name: str) -> dict[str, str | None] | None:
username = self._config.get(f"http-basic.{repo_name}.username")
password = self._config.get(f"http-basic.{repo_name}.password")
if not username and not password:
return None

if not password:
password = self.keyring.get_password(name, username)
if self.use_keyring:
password = self.keyring.get_password(repo_name, username)
else:
return None

return {
"username": username,
"password": password,
}

def set_http_password(self, name: str, username: str, password: str) -> None:
def set_http_password(self, repo_name: str, username: str, password: str) -> None:
auth = {"username": username}

if not self.keyring.is_available():
if not self.use_keyring:
self.warn_plaintext_credentials_stored()
auth["password"] = password
else:
self.keyring.set_password(name, username, password)
self.keyring.set_password(repo_name, username, password)

self._config.auth_config_source.add_property(f"http-basic.{name}", auth)
self._config.auth_config_source.add_property(f"http-basic.{repo_name}", auth)

def delete_http_password(self, name: str) -> None:
auth = self.get_http_auth(name)
def delete_http_password(self, repo_name: str) -> None:
auth = self.get_http_auth(repo_name)
if not auth:
return

Expand All @@ -237,6 +234,14 @@ def delete_http_password(self, name: str) -> None:
return

with suppress(PoetryKeyringError):
self.keyring.delete_password(name, username)
self.keyring.delete_password(repo_name, username)

self._config.auth_config_source.remove_property(f"http-basic.{repo_name}")

self._config.auth_config_source.remove_property(f"http-basic.{name}")
def get_credential(
self, *names: str, username: str | None = None
) -> HTTPAuthCredential:
if self.use_keyring:
return self.keyring.get_credential(*names, username=username)
else:
return HTTPAuthCredential(username=username, password=None)
Loading

0 comments on commit 753ab64

Please sign in to comment.