__author__ = "Vanessa Sochat"
__copyright__ = "Copyright The ORAS Authors."
__license__ = "Apache-2.0"
import sys
from typing import List, Optional, Union
import oras.auth
import oras.container
import oras.main.login as login
import oras.provider
import oras.utils
import oras.version
[docs]
class OrasClient:
"""
Create an OCI Registry as Storage (ORAS) Client.
This is intended for controlled interactions. The user of oras-py can use
this client, the terminal command line wrappers, or the functions in main
in isolation as an internal Python API. The user can provide a custom
registry as a parameter, if desired. If not provided we default to standard
oras.
"""
def __init__(
self,
hostname: Optional[str] = None,
registry: Optional[oras.provider.Registry] = None,
insecure: bool = False,
tls_verify: bool = True,
):
"""
Create an ORAS client.
The hostname is the remote registry to ping.
:param hostname: the hostname of the registry to ping
:type hostname: str
:param registry: if provided, use this custom provider instead of default
:type registry: oras.provider.Registry or None
:param insecure: use http instead of https
:type insecure: bool
"""
self.remote = registry or oras.provider.Registry(hostname, insecure, tls_verify)
def __repr__(self) -> str:
return str(self)
def __str__(self) -> str:
return "[oras-client]"
[docs]
def set_token_auth(self, token: str):
"""
Set token authentication.
:param token: the bearer token
:type token: str
"""
self.remote.set_token_auth(token)
[docs]
def set_basic_auth(self, username: str, password: str):
"""
Add basic authentication to the request.
:param username: the user account name
:type username: str
:param password: the user account password
:type password: str
"""
self.remote.set_basic_auth(username, password)
[docs]
def version(self, return_items: bool = False) -> Union[dict, str]:
"""
Get the version of the client.
:param return_items : return the dict of version info instead of string
:type return_items: bool
"""
version = oras.version.__version__
python_version = "%s.%s.%s" % (
sys.version_info.major,
sys.version_info.minor,
sys.version_info.micro,
)
versions = {"Version": version, "Python version": python_version}
# If the user wants the dictionary of items returned
if return_items:
return versions
# Otherwise return a string that can be printed
return "\n".join(["%s: %s" % (k, v) for k, v in versions.items()])
[docs]
def push(self, *args, **kwargs):
"""
Push a container to the remote.
"""
return self.remote.push(*args, **kwargs)
[docs]
def pull(self, *args, **kwargs):
"""
Pull a container from the remote.
"""
return self.remote.pull(*args, **kwargs)
[docs]
def login(
self,
username: str,
password: str,
password_stdin: bool = False,
insecure: bool = False,
tls_verify: bool = True,
hostname: Optional[str] = None,
config_path: Optional[List[str]] = None,
) -> dict:
"""
Login to a registry.
:param registry: if provided, use this custom provider instead of default
:type registry: oras.provider.Registry or None
:param username: the user account name
:type username: str
:param password: the user account password
:type password: str
:param password_stdin: get the password from standard input
:type password_stdin: bool
:param insecure: use http instead of https
:type insecure: bool
:param tls_verify: verify tls
:type tls_verify: bool
:param hostname: the hostname to login to
:type hostname: str
:param config_path: list of config paths to add
:type config_path: list
"""
login_func = self._login
if hasattr(self.remote, "login"):
login_func = self.remote.login # type: ignore
return login_func(
username=username,
password=password,
password_stdin=password_stdin,
tls_verify=tls_verify,
hostname=hostname,
config_path=config_path, # type: ignore
)
[docs]
def logout(self, hostname: str):
"""
Logout from a registry, meaning removing any auth (if loaded)
:param hostname: the hostname to login to
:type hostname: str
"""
self.remote.logout(hostname)
def _login(
self,
username: Optional[str] = None,
password: Optional[str] = None,
password_stdin: bool = False,
tls_verify: bool = True,
hostname: Optional[str] = None,
config_path: Optional[str] = None,
) -> dict:
"""
Login to an OCI registry.
The username and password can come from stdin. Most people use username
password to get a token, so we are disabling providing just a token for
now. A tool that wants to provide a token should use set_token_auth.
"""
# Read password from stdin
if password_stdin:
password = oras.utils.readline()
# No username, try to get from stdin
if not username:
username = input("Username: ")
# No password provided
if not password:
password = input("Password: ")
if not password:
raise ValueError("password required")
# Cut out early if we didn't get what we need
if not password or not username:
return {"Login": "Not successful"}
# Set basic auth for the client
self.set_basic_auth(username, password)
# Login
# https://docker-py.readthedocs.io/en/stable/client.html?highlight=login#docker.client.DockerClient.login
try:
client = oras.utils.get_docker_client(tls_verify=tls_verify)
return client.login(
username=username,
password=password,
registry=hostname,
dockercfg_path=config_path,
)
# Fallback to manual login
except Exception:
return login.DockerClient().login(
username=username, # type: ignore
password=password, # type: ignore
registry=hostname, # type: ignore
dockercfg_path=config_path,
)