File: //proc/thread-self/root/usr/lib/python3/dist-packages/uaclient/contract.py
import logging
from typing import Any, Dict, List, Optional
from uaclient import clouds, exceptions, serviceclient, status, util
from uaclient.config import UAConfig
API_V1_CONTEXT_MACHINE_TOKEN = "/v1/context/machines/token"
API_V1_TMPL_CONTEXT_MACHINE_TOKEN_RESOURCE = (
"/v1/contracts/{contract}/context/machines/{machine}"
)
API_V1_RESOURCES = "/v1/resources"
API_V1_TMPL_RESOURCE_MACHINE_ACCESS = (
"/v1/resources/{resource}/context/machines/{machine}"
)
API_V1_AUTO_ATTACH_CLOUD_TOKEN = "/v1/clouds/{cloud_type}/token"
API_V1_MACHINE_ACTIVITY = "/v1/contracts/{contract}/machine-activity/{machine}"
API_V1_CONTRACT_INFORMATION = "/v1/contract"
ATTACH_FAIL_DATE_FORMAT = "%B %d, %Y"
class ContractAPIError(util.UrlError):
def __init__(self, e, error_response):
super().__init__(e, e.code, e.headers, e.url)
if "error_list" in error_response:
self.api_errors = error_response["error_list"]
else:
self.api_errors = [error_response]
for error in self.api_errors:
error["code"] = error.get("title", error.get("code"))
def __contains__(self, error_code):
for error in self.api_errors:
if error_code == error.get("code"):
return True
if error.get("message", "").startswith(error_code):
return True
return False
def __get__(self, error_code, default=None):
for error in self.api_errors:
if error["code"] == error_code:
return error["detail"]
return default
def __str__(self):
prefix = super().__str__()
details = []
for err in self.api_errors:
if not err.get("extra"):
details.append(err.get("detail", err.get("message", "")))
else:
for extra in err["extra"].values():
if isinstance(extra, list):
details.extend(extra)
else:
details.append(extra)
return prefix + ": [" + self.url + "]" + ", ".join(details)
class UAContractClient(serviceclient.UAServiceClient):
cfg_url_base_attr = "contract_url"
api_error_cls = ContractAPIError
def request_contract_machine_attach(self, contract_token, machine_id=None):
"""Requests machine attach to the provided contact_id.
@param contract_id: Unique contract id provided by contract service.
@param contract_token: Token string providing authentication to
ContractBearer service endpoint.
@param machine_id: Optional unique system machine id. When absent,
contents of /etc/machine-id will be used.
@return: Dict of the JSON response containing the machine-token.
"""
headers = self.headers()
headers.update({"Authorization": "Bearer {}".format(contract_token)})
data = self._get_platform_data(machine_id)
machine_token, _headers = self.request_url(
API_V1_CONTEXT_MACHINE_TOKEN, data=data, headers=headers
)
self.cfg.write_cache("machine-token", machine_token)
util.get_machine_id.cache_clear()
machine_id = machine_token.get("machineTokenInfo", {}).get(
"machineId", data.get("machineId")
)
self.cfg.write_cache("machine-id", machine_id)
return machine_token
def request_resources(self) -> Dict[str, Any]:
"""Requests list of entitlements available to this machine type."""
platform = util.get_platform_info()
query_params = {
"architecture": platform["arch"],
"series": platform["series"],
"kernel": platform["kernel"],
}
resource_response, headers = self.request_url(
API_V1_RESOURCES, query_params=query_params
)
return resource_response
def request_contract_information(
self, contract_token: str
) -> Dict[str, Any]:
headers = self.headers()
headers.update({"Authorization": "Bearer {}".format(contract_token)})
response_data, _response_headers = self.request_url(
API_V1_CONTRACT_INFORMATION, headers=headers
)
return response_data
def request_auto_attach_contract_token(
self, *, instance: clouds.AutoAttachCloudInstance
):
"""Requests contract token for auto-attach images for Pro clouds.
@param instance: AutoAttachCloudInstance for the cloud.
@return: Dict of the JSON response containing the contract-token.
"""
response, _headers = self.request_url(
API_V1_AUTO_ATTACH_CLOUD_TOKEN.format(
cloud_type=instance.cloud_type
),
data=instance.identity_doc,
)
self.cfg.write_cache("contract-token", response)
return response
def request_resource_machine_access(
self,
machine_token: str,
resource: str,
machine_id: Optional[str] = None,
) -> Dict[str, Any]:
"""Requests machine access context for a given resource
@param machine_token: The authentication token needed to talk to
this contract service endpoint.
@param resource: Entitlement name.
@param machine_id: Optional unique system machine id. When absent,
contents of /etc/machine-id will be used.
@return: Dict of the JSON response containing entitlement accessInfo.
"""
if not machine_id:
machine_id = util.get_machine_id(self.cfg)
headers = self.headers()
headers.update({"Authorization": "Bearer {}".format(machine_token)})
url = API_V1_TMPL_RESOURCE_MACHINE_ACCESS.format(
resource=resource, machine=machine_id
)
resource_access, headers = self.request_url(url, headers=headers)
if headers.get("expires"):
resource_access["expires"] = headers["expires"]
self.cfg.write_cache(
"machine-access-{}".format(resource), resource_access
)
return resource_access
def request_machine_token_update(
self, machine_token: str, contract_id: str, machine_id: str = None
) -> Dict:
"""Update existing machine-token for an attached machine."""
return self._request_machine_token_update(
machine_token=machine_token,
contract_id=contract_id,
machine_id=machine_id,
detach=False,
)
def report_machine_activity(self, enabled_services: "List[str]"):
"""Report current activity token and enabled services.
This will report to the contracts backend all the current
enabled services in the system.
"""
contract_id = self.cfg.contract_id
activity_token = self.cfg.activity_token
machine_token = self.cfg.machine_token.get("machineToken")
machine_id = util.get_machine_id(self.cfg)
# If the activityID is null we should provide the endpoint
# with the instance machine id as the activityID
activity_id = self.cfg.activity_id or machine_id
request_data = {
"activityToken": activity_token,
"activityID": activity_id,
"resources": enabled_services,
}
url = API_V1_MACHINE_ACTIVITY.format(
contract=contract_id, machine=machine_id
)
headers = self.headers()
headers.update({"Authorization": "Bearer {}".format(machine_token)})
response, _ = self.request_url(url, headers=headers, data=request_data)
# We will update the `machine-token.json` based on the response
# provided by the server. We expect the response to be
# a full `activityInfo` object which belongs at the root of
# `machine-token.json`
if response:
machine_token = self.cfg.read_cache("machine-token")
# The activity information received as a response here
# will not provide the information inside an activityInfo
# structure. However, this structure will be reflected when
# we reach the contract for attach and refresh requests.
# Because of that, we will store the response directly on
# the activityInfo key
machine_token["activityInfo"] = response
self.cfg.write_cache("machine-token", machine_token)
def detach_machine_from_contract(
self, machine_token: str, contract_id: str, machine_id: str = None
) -> Dict:
"""Report the attached machine should be detached from the contract."""
curr_machine_id = self._get_platform_data(machine_id=None).get(
"machineId", ""
)
past_machine_id = self.cfg.read_cache("machine-id")
if str(curr_machine_id) != str(past_machine_id):
logging.debug(
"Found new machine-id. Do not call detach on contract backend"
)
return {}
return self._request_machine_token_update(
machine_token=machine_token,
contract_id=contract_id,
machine_id=machine_id,
detach=True,
)
def _request_machine_token_update(
self,
machine_token: str,
contract_id: str,
machine_id: str = None,
detach: bool = False,
) -> Dict:
"""Request machine token refresh from contract server.
@param machine_token: The machine token needed to talk to
this contract service endpoint.
@param contract_id: Unique contract id provided by contract service.
@param machine_id: Optional unique system machine id. When absent,
contents of /etc/machine-id will be used.
@param detach: Boolean set True if detaching this machine from the
active contract. Default is False.
@return: Dict of the JSON response containing refreshed machine-token
"""
headers = self.headers()
headers.update({"Authorization": "Bearer {}".format(machine_token)})
data = self._get_platform_data(machine_id)
url = API_V1_TMPL_CONTEXT_MACHINE_TOKEN_RESOURCE.format(
contract=contract_id, machine=data["machineId"]
)
kwargs = {"headers": headers}
if detach:
kwargs["method"] = "DELETE"
else:
kwargs["method"] = "POST"
kwargs["data"] = data
response, headers = self.request_url(url, **kwargs)
if headers.get("expires"):
response["expires"] = headers["expires"]
if not detach:
self.cfg.write_cache("machine-token", response)
util.get_machine_id.cache_clear()
machine_id = response.get("machineTokenInfo", {}).get(
"machineId", data.get("machineId")
)
self.cfg.write_cache("machine-id", machine_id)
return response
def _get_platform_data(self, machine_id):
"""Return a dict of platform-related data for contract requests"""
if not machine_id:
machine_id = util.get_machine_id(self.cfg)
platform = util.get_platform_info()
platform_os = platform.copy()
arch = platform_os.pop("arch")
return {
"machineId": machine_id,
"architecture": arch,
"os": platform_os,
}
def process_entitlements_delta(
past_entitlements: Dict[str, Any],
new_entitlements: Dict[str, Any],
allow_enable: bool,
series_overrides: bool = True,
) -> None:
"""Iterate over all entitlements in new_entitlement and apply any delta
found according to past_entitlements.
:param past_entitlements: dict containing the last valid information
regarding service entitlements.
:param new_entitlements: dict containing the current information regarding
service entitlements.
:param allow_enable: Boolean set True if allowed to perform the enable
operation. When False, a message will be logged to inform the user
about the recommended enabled service.
:param series_overrides: Boolean set True if series overrides should be
applied to the new_access dict.
"""
delta_error = False
unexpected_error = False
for name, new_entitlement in sorted(new_entitlements.items()):
try:
process_entitlement_delta(
past_entitlements.get(name, {}),
new_entitlement,
allow_enable=allow_enable,
series_overrides=series_overrides,
)
except exceptions.UserFacingError:
delta_error = True
with util.disable_log_to_console():
logging.error(
"Failed to process contract delta for {name}:"
" {delta}".format(name=name, delta=new_entitlement)
)
except Exception:
unexpected_error = True
with util.disable_log_to_console():
logging.exception(
"Unexpected error processing contract delta for {name}:"
" {delta}".format(name=name, delta=new_entitlement)
)
if unexpected_error:
raise exceptions.UserFacingError(status.MESSAGE_UNEXPECTED_ERROR)
elif delta_error:
raise exceptions.UserFacingError(
status.MESSAGE_ATTACH_FAILURE_DEFAULT_SERVICES
)
def process_entitlement_delta(
orig_access: Dict[str, Any],
new_access: Dict[str, Any],
allow_enable: bool = False,
series_overrides: bool = True,
) -> Dict:
"""Process a entitlement access dictionary deltas if they exist.
:param orig_access: Dict with original entitlement access details before
contract refresh deltas
:param new_access: Dict with updated entitlement access details after
contract refresh
:param allow_enable: Boolean set True if allowed to perform the enable
operation. When False, a message will be logged to inform the user
about the recommended enabled service.
:param series_overrides: Boolean set True if series overrides should be
applied to the new_access dict.
:raise UserFacingError: on failure to process deltas.
:return: Dict of processed deltas
"""
from uaclient.entitlements import entitlement_factory
if series_overrides:
util.apply_series_overrides(new_access)
deltas = util.get_dict_deltas(orig_access, new_access)
if deltas:
name = orig_access.get("entitlement", {}).get("type")
if not name:
name = deltas.get("entitlement", {}).get("type")
if not name:
raise RuntimeError(
"Could not determine contract delta service type {} {}".format(
orig_access, new_access
)
)
ent_cls = entitlement_factory(name)
if not ent_cls:
logging.debug(
'Skipping entitlement deltas for "%s". No such class', name
)
return deltas
entitlement = ent_cls(assume_yes=allow_enable)
entitlement.process_contract_deltas(
orig_access, deltas, allow_enable=allow_enable
)
return deltas
def _create_attach_forbidden_message(e: ContractAPIError) -> str:
msg = status.MESSAGE_ATTACH_EXPIRED_TOKEN
if (
hasattr(e, "api_errors")
and len(e.api_errors) > 0
and "info" in e.api_errors[0]
):
info = e.api_errors[0]["info"]
contract_id = info["contractId"]
reason = info["reason"]
reason_msg = ""
if reason == "no-longer-effective":
date = info["time"].strftime(ATTACH_FAIL_DATE_FORMAT)
reason_msg = status.MESSAGE_ATTACH_FORBIDDEN_EXPIRED.format(
contract_id=contract_id, date=date
)
elif reason == "not-effective-yet":
date = info["time"].strftime(ATTACH_FAIL_DATE_FORMAT)
reason_msg = status.MESSAGE_ATTACH_FORBIDDEN_NOT_YET.format(
contract_id=contract_id, date=date
)
elif reason == "never-effective":
reason_msg = status.MESSAGE_ATTACH_FORBIDDEN_NEVER.format(
contract_id=contract_id
)
msg = status.MESSAGE_ATTACH_FORBIDDEN.format(reason=reason_msg)
return msg
def request_updated_contract(
cfg, contract_token: Optional[str] = None, allow_enable=False
):
"""Request contract refresh from ua-contracts service.
Compare original token to new token and react to entitlement deltas.
:param cfg: Instance of UAConfig for this machine.
:param contract_token: String contraining an optional contract token.
:param allow_enable: Boolean set True if allowed to perform the enable
operation. When False, a message will be logged to inform the user
about the recommended enabled service.
:raise UserFacingError: on failure to update contract or error processing
contract deltas
:raise UrlError: On failure to contact the server
"""
orig_token = cfg.machine_token
orig_entitlements = cfg.entitlements
if orig_token and contract_token:
raise RuntimeError(
"Got unexpected contract_token on an already attached machine"
)
contract_client = UAContractClient(cfg)
if contract_token: # We are a mid ua-attach and need to get machinetoken
try:
contract_client.request_contract_machine_attach(
contract_token=contract_token
)
except util.UrlError as e:
if isinstance(e, ContractAPIError):
if hasattr(e, "code"):
if e.code == 401:
raise exceptions.UserFacingError(
status.MESSAGE_ATTACH_INVALID_TOKEN
)
elif e.code == 403:
msg = _create_attach_forbidden_message(e)
raise exceptions.UserFacingError(msg)
raise e
with util.disable_log_to_console():
logging.exception(str(e))
raise exceptions.UserFacingError(status.MESSAGE_CONNECTIVITY_ERROR)
else:
machine_token = orig_token["machineToken"]
contract_id = orig_token["machineTokenInfo"]["contractInfo"]["id"]
contract_client.request_machine_token_update(
machine_token=machine_token, contract_id=contract_id
)
process_entitlements_delta(
orig_entitlements, cfg.entitlements, allow_enable
)
def get_available_resources(cfg: UAConfig) -> List[Dict]:
"""Query available resources from the contract server for this machine."""
client = UAContractClient(cfg)
resources = client.request_resources()
return resources.get("resources", [])
def get_contract_information(cfg: UAConfig, token: str) -> Dict[str, Any]:
"""Query contract information for a specific token"""
client = UAContractClient(cfg)
return client.request_contract_information(token)