Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Better authentication handling in backends #215

Merged
merged 20 commits into from
Apr 19, 2024
Merged
Show file tree
Hide file tree
Changes from 16 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
228 changes: 115 additions & 113 deletions audbackend/core/backend/artifactory.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,46 +10,6 @@
from audbackend.core.backend.base import Base


def _artifactory_path(
path,
username,
apikey,
) -> artifactory.ArtifactoryPath:
r"""Authenticate at Artifactory and get path object."""
return artifactory.ArtifactoryPath(
path,
auth=(username, apikey),
)


def _authentication(host) -> typing.Tuple[str, str]:
"""Look for username and API key."""
username = os.getenv("ARTIFACTORY_USERNAME", None)
api_key = os.getenv("ARTIFACTORY_API_KEY", None)
config_file = os.getenv(
"ARTIFACTORY_CONFIG_FILE",
artifactory.default_config_path,
)
config_file = audeer.path(config_file)

if os.path.exists(config_file) and (api_key is None or username is None):
config = artifactory.read_config(config_file)
config_entry = artifactory.get_config_entry(config, host)

if config_entry is not None:
if username is None:
username = config_entry.get("username", None)
if api_key is None:
api_key = config_entry.get("password", None)

if username is None:
username = "anonymous"
if api_key is None:
api_key = ""

return username, api_key


def _deploy(
src_path: str,
dst_path: artifactory.ArtifactoryPath,
Expand Down Expand Up @@ -109,45 +69,106 @@ def _download(
class Artifactory(Base):
r"""Backend for Artifactory.

Looks for the two environment variables
``ARTIFACTORY_USERNAME`` and
``ARTIFACTORY_API_KEY``.
Otherwise,
tries to extract missing values
from a global `config file`_.
The default path of the config file
(:file:`~/.artifactory_python.cfg`)
can be overwritten with the environment variable
``ARTIFACTORY_CONFIG_FILE``.
If no config file exists
or if it does not contain an
entry for the ``host``,
the username is set to ``'anonymous'``
and the API key to an empty string.
In that case the ``host``
should support anonymous access.

Args:
host: host address
repository: repository name

.. _`config file`: https://devopshq.github.io/artifactory/#global-configuration-file
auth: username, password / API key / access token tuple.
If ``None``,
it requests it by calling :meth:`authentication`

""" # noqa: E501

def __init__(
self,
host,
repository,
host: str,
repository: str,
*,
auth: typing.Tuple = None,
hagenw marked this conversation as resolved.
Show resolved Hide resolved
):
super().__init__(host, repository)
super().__init__(host, repository, auth=auth)

if auth is None:
self.auth = self.authentication(host)

# Store ArtifactoryPath object to the repository,
# when opening the backend.
self._repo = None

@classmethod
def authentication(cls, host: str) -> typing.Tuple[str, str]:
"""Username and password/access token for given host.

Returns a username
and password / API key / access token,
which can be used to authenticate
with an Artifactory server.

Note, API keys are deprecated
and will no longer work
with newer versions of Artifactory.

To get the username,
password/access token combination,
the function looks first
for the two environment variables
``ARTIFACTORY_USERNAME`` and
``ARTIFACTORY_API_KEY``.
frankenjoe marked this conversation as resolved.
Show resolved Hide resolved
Otherwise,
it tries to extract missing values
from a global `config file`_.
The default path of the config file
(:file:`~/.artifactory_python.cfg`)
can be overwritten with the environment variable
``ARTIFACTORY_CONFIG_FILE``.
If no config file exists
or if it does not contain an
entry for the ``host``,
the username is set to ``'anonymous'``
and the password/key to an empty string.
In that case the ``host``
has to support anonymous access,
when trying to authenticate.

.. _`config file`: https://devopshq.github.io/artifactory/#global-configuration-file

Args:
host: hostname of Artifactory backend

Returns:
username, password / API key / access token tuple

"""
username = os.getenv("ARTIFACTORY_USERNAME", None)
api_key = os.getenv("ARTIFACTORY_API_KEY", None)
config_file = os.getenv(
"ARTIFACTORY_CONFIG_FILE",
artifactory.default_config_path,
)
config_file = audeer.path(config_file)

if os.path.exists(config_file) and (api_key is None or username is None):
config = artifactory.read_config(config_file)
config_entry = artifactory.get_config_entry(config, host)

if config_entry is not None:
if username is None:
username = config_entry.get("username", None)
if api_key is None:
api_key = config_entry.get("password", None)

if username is None:
username = "anonymous"
if api_key is None:
api_key = ""

return username, api_key

def _checksum(
self,
path: str,
) -> str:
r"""MD5 checksum of file on backend."""
path = self._path(path)
path = self.path(path)
checksum = artifactory.ArtifactoryPath.stat(path).md5
return checksum

Expand All @@ -162,7 +183,7 @@ def _collapse(
/<path>

"""
path = path[len(str(self._repo.path)) - 1 :]
path = path[len(str(self.path("/"))) - 1 :]
path = path.replace("/", self.sep)
return path

Expand All @@ -173,8 +194,8 @@ def _copy_file(
verbose: bool,
):
r"""Copy file on backend."""
src_path = self._path(src_path)
dst_path = self._path(dst_path)
src_path = self.path(src_path)
dst_path = self.path(dst_path)
if not dst_path.parent.exists():
dst_path.parent.mkdir()
src_path.copy(dst_path)
Expand All @@ -183,8 +204,7 @@ def _create(
self,
):
r"""Access existing repository."""
username, api_key = _authentication(self.host)
path = _artifactory_path(self.host, username, api_key)
path = artifactory.ArtifactoryPath(self.host, auth=self.auth)
repo = dohq_artifactory.RepositoryLocal(
path,
self.repository,
Expand All @@ -199,7 +219,7 @@ def _date(
path: str,
) -> str:
r"""Get last modification date of file on backend."""
path = self._path(path)
path = self.path(path)
date = path.stat().mtime
date = utils.date_format(date)
return date
Expand All @@ -216,42 +236,25 @@ def _exists(
path: str,
) -> bool:
r"""Check if file exists on backend."""
path = self._path(path)
path = self.path(path)
return path.exists()

def _expand(
self,
path: str,
) -> str:
r"""Convert to backend path.

<path>
->
<host>/<repository>/<path>

"""
path = path.replace(self.sep, "/")
if path.startswith("/"):
path = path[1:]
path = f"{self._repo.path}{path}"
return path

def _get_file(
self,
src_path: str,
dst_path: str,
verbose: bool,
):
r"""Get file from backend."""
src_path = self._path(src_path)
src_path = self.path(src_path)
_download(src_path, dst_path, verbose=verbose)

def _ls(
self,
path: str,
) -> typing.List[str]:
r"""List all files under sub-path."""
path = self._path(path)
path = self.path(path)
if not path.exists():
return []

Expand All @@ -267,8 +270,8 @@ def _move_file(
verbose: bool,
):
r"""Move file on backend."""
src_path = self._path(src_path)
dst_path = self._path(dst_path)
src_path = self.path(src_path)
dst_path = self.path(dst_path)
if not dst_path.parent.exists():
dst_path.parent.mkdir()
src_path.move(dst_path)
Expand All @@ -277,13 +280,8 @@ def _open(
self,
):
r"""Open connection to backend."""
self._username, self._api_key = _authentication(self.host)
path = _artifactory_path(
self.host,
self._username,
self._api_key,
)
self._repo = path.find_repository_local(self.repository)
path = artifactory.ArtifactoryPath(self.host, auth=self.auth)
self._repo = path.find_repository(self.repository)
frankenjoe marked this conversation as resolved.
Show resolved Hide resolved
if self._repo is None:
utils.raise_file_not_found_error(self.repository)

Expand All @@ -292,28 +290,32 @@ def _owner(
path: str,
) -> str:
r"""Get owner of file on backend."""
path = self._path(path)
path = self.path(path)
owner = path.stat().modified_by
return owner

def _path(
def path(
self,
path: str,
) -> artifactory.ArtifactoryPath:
r"""Convert to backend path.

<path>
->
<host>/<repository>/<path>
This extends the relative ``path`` on the backend
by :attr:`host` and :attr:`repository`,
and returns an :class:`artifactory.ArtifactortPath` object.
hagenw marked this conversation as resolved.
Show resolved Hide resolved

Args:
path: path on backend

Returns:
Artifactory path object

"""
path = self._expand(path)
path = _artifactory_path(
path,
self._username,
self._api_key,
)
return path
path = path.replace(self.sep, "/")
if path.startswith("/"):
path = path[1:]
# path -> host/repository/path
return self._repo / path

def _put_file(
self,
Expand All @@ -323,13 +325,13 @@ def _put_file(
verbose: bool,
):
r"""Put file to backend."""
dst_path = self._path(dst_path)
dst_path = self.path(dst_path)
_deploy(src_path, dst_path, checksum, verbose=verbose)

def _remove_file(
self,
path: str,
):
r"""Remove file from backend."""
path = self._path(path)
path = self.path(path)
path.unlink()
Loading