File manager - Edit - /home/newsbmcs.com/public_html/static/img/logo/cli.tar
Back
__pycache__/__init__.cpython-310.pyc 0000644 00000010746 15030124134 0013215 0 ustar 00 o �h2 � @ s� d Z ddlmZ ddlZddlZddlZddlmZ G dd� de�Zd d d �Z d d d�Z d dd�Z d dd�Z d dd�Z d dd�ZdZdd� Zd d � ZdS ) z Wheel command-line utility. � )� annotationsN)�ArgumentTypeErrorc @ s e Zd ZdS )� WheelErrorN)�__name__� __module__� __qualname__� r r �H/usr/local/CyberPanel/lib/python3.10/site-packages/wheel/cli/__init__.pyr s r �args�argparse.Namespace�return�Nonec C s ddl m } || j| j� d S )N� )�unpack)r � wheelfile�dest)r r r r r �unpack_f � r c C �" ddl m } || j| j| j� d S )Nr )�pack)r � directory�dest_dir� build_number)r r r r r �pack_f � r c C r )Nr )�convert)r �filesr �verbose)r r r r r � convert_f r r c s8 ddl m � � �fdd�� jD �}|D ]}t|� qd S )Nr )�tagsc 3 s, � | ]}�|� j � j� j� j� j�V qd S �N)� python_tag�abi_tag� platform_tag�build�remove)�.0�wheel�r r r r � <genexpr>& s � �� �ztags_f.<locals>.<genexpr>)r r' �print)r �names�namer r( r �tags_f# s � �r- c C s ddl m} td|� �� d S )N� )� __version__zwheel )� r/ r* )r r/ r r r � version_f6 r r1 � build_tag�strc C s, | r | d � � s td��d| v rtd��| S )Nr z!build tag must begin with a digit�-z$invalid character ('-') in build tag)�isdigitr )r2 r r r �parse_build_tag< s r6 ay Make a new wheel with given tags. Any tags unspecified will remain the same. Starting the tags with a "+" will append to the existing tags. Starting with a "-" will remove a tag (use --option=-TAG syntax). Multiple tags can be separated by ".". The original file will remain unless --remove is given. The output filename(s) will be displayed on stdout for further processing. c s� t �� � � jdd�} | jddd�}|jddddd � |jd d d� |jtd � | jd dd�}|jddd� |jddtjj dd� |jddd� |jt d � | jddd�}|jdddd� |jddtjj d d� |jd d d d � |jt d � | jd!d"t d#�}|jd$dd%d� |jd&d d'd(� |jd)d*d+d,� |jd-d*d.d,� |jd/d*d0d,� |jd1t d2d3d4� |jtd � | jd5d6d�}|jtd � | jd7d8d�}|j� fd9d:�d � � S );N�commands)�helpr z Unpack wheelz--destz-dzDestination directory�.)r8 �defaultr z Wheel file)�funcr z Repack wheelr z$Root directory of the unpacked wheelz --dest-dirz2Directory to store the wheel (default %(default)s))r: r8 z--build-numberz"Build tag to use in the wheel namer z Convert egg or wininst to wheelr �*zFiles to convert)�nargsr8 z/Directory to store wheels (default %(default)s)z --verbosez-v� store_true)�actionr z"Add or replace the tags on a wheel)r8 � descriptionr' zExisting wheel(s) to retagz--removez8Remove the original files, keeping only the renamed ones)r? r8 z --python-tag�TAGz Specify an interpreter tag(s))�metavarr8 z --abi-tagzSpecify an ABI tag(s)z--platform-tagzSpecify a platform tag(s)z--build�BUILDzSpecify a build tag)�typerB r8 �versionzPrint version and exitr8 zShow this helpc s � � � S r )� print_help)r ��pr r �<lambda>� s zparser.<locals>.<lambda>)�argparse�ArgumentParser�add_subparsers� add_parser� add_argument� set_defaultsr �os�path�curdirr r � TAGS_HELPr6 r- r1 )�s� unpack_parser� repack_parser�convert_parser� tags_parser�version_parser� help_parserr rG r �parserN sn � �� � ����� r[ c C sj t � } | �� }t|d�s| �� dS z|�|� W dS ty4 } zt|tjd� W Y d }~dS d }~ww )Nr; r )�filer ) r[ � parse_args�hasattrrF r; r r* �sys�stderr)rH r �er r r �main� s � ��rb )r r r r )r2 r3 r r3 )�__doc__� __future__r rJ rP r_ r � Exceptionr r r r r- r1 r6 rS r[ rb r r r r �<module> s A __pycache__/chardetect.cpython-310.pyc 0000644 00000005062 15030124134 0013557 0 ustar 00 o �n�_� � @ sv d Z ddlmZmZmZ ddlZddlZddlmZ ddl m Z ddl m Z d dd �Z dd d �Zed kr9e� dS dS )a Script which takes one or more file paths and reports on their detected encodings Example:: % chardetect somefile someotherfile somefile: windows-1252 with confidence 0.5 someotherfile: ascii with confidence 1.0 If no paths are provided, it takes its input from stdin. � )�absolute_import�print_function�unicode_literalsN)� __version__)�PY2)�UniversalDetector�stdinc C sv t � }| D ]}t|�}|�|� |jr nq|�� |j}tr'|�t� � d�}|d r6d� ||d |d �S d� |�S )z� Return a string describing the probable encoding of a file or list of strings. :param lines: The lines to get the encoding of. :type lines: Iterable of bytes :param name: Name of file or collection of lines :type name: str �ignore�encodingz{}: {} with confidence {}� confidencez {}: no result) r � bytearray�feed�done�close�resultr �decode�sys�getfilesystemencoding�format)�lines�name�u�liner � r �8/usr/lib/python3/dist-packages/chardet/cli/chardetect.py�description_of s � � r c C s� t jdd�}|jddt �d�dtrtjntjjgd� |jdd d �t �d � |� | �}|j D ]}|� � r;t d tjd � t t||j�� q.dS )z� Handles command line arguments and gets things started. :param argv: List of arguments, as if specified on the command-line. If None, ``sys.argv[1:]`` is used instead. :type argv: list of str zVTakes one or more file paths and reports their detected encodings)� description�inputz^File whose encoding we would like to determine. (default: stdin)�rb�*)�help�type�nargs�defaultz --version�versionz %(prog)s {})�actionr$ z�You are running chardetect interactively. Press CTRL-D twice at the start of a blank line to signal the end of your input. If you want help, run chardetect --help )�fileN)�argparse�ArgumentParser� add_argument�FileTyper r r �bufferr r � parse_argsr �isatty�print�stderrr r )�argv�parser�args�fr r r �main5 s& � �� ��r4 �__main__)r )N)�__doc__� __future__r r r r' r �chardetr �chardet.compatr �chardet.universaldetectorr r r4 �__name__r r r r �<module> s � chardetect.py 0000644 00000005227 15030124134 0007223 0 ustar 00 """ Script which takes one or more file paths and reports on their detected encodings Example:: % chardetect somefile someotherfile somefile: windows-1252 with confidence 0.5 someotherfile: ascii with confidence 1.0 If no paths are provided, it takes its input from stdin. """ from __future__ import absolute_import, print_function, unicode_literals import argparse import sys from chardet import __version__ from chardet.compat import PY2 from chardet.universaldetector import UniversalDetector def description_of(lines, name='stdin'): """ Return a string describing the probable encoding of a file or list of strings. :param lines: The lines to get the encoding of. :type lines: Iterable of bytes :param name: Name of file or collection of lines :type name: str """ u = UniversalDetector() for line in lines: line = bytearray(line) u.feed(line) # shortcut out of the loop to save reading further - particularly useful if we read a BOM. if u.done: break u.close() result = u.result if PY2: name = name.decode(sys.getfilesystemencoding(), 'ignore') if result['encoding']: return '{}: {} with confidence {}'.format(name, result['encoding'], result['confidence']) else: return '{}: no result'.format(name) def main(argv=None): """ Handles command line arguments and gets things started. :param argv: List of arguments, as if specified on the command-line. If None, ``sys.argv[1:]`` is used instead. :type argv: list of str """ # Get command line arguments parser = argparse.ArgumentParser( description="Takes one or more file paths and reports their detected \ encodings") parser.add_argument('input', help='File whose encoding we would like to determine. \ (default: stdin)', type=argparse.FileType('rb'), nargs='*', default=[sys.stdin if PY2 else sys.stdin.buffer]) parser.add_argument('--version', action='version', version='%(prog)s {}'.format(__version__)) args = parser.parse_args(argv) for f in args.input: if f.isatty(): print("You are running chardetect interactively. Press " + "CTRL-D twice at the start of a blank line to signal the " + "end of your input. If you want help, run chardetect " + "--help\n", file=sys.stderr) print(description_of(f, f.name)) if __name__ == '__main__': main() __init__.py 0000644 00000010462 15030124134 0006651 0 ustar 00 """ Wheel command-line utility. """ from __future__ import annotations import argparse import os import sys from argparse import ArgumentTypeError class WheelError(Exception): pass def unpack_f(args: argparse.Namespace) -> None: from .unpack import unpack unpack(args.wheelfile, args.dest) def pack_f(args: argparse.Namespace) -> None: from .pack import pack pack(args.directory, args.dest_dir, args.build_number) def convert_f(args: argparse.Namespace) -> None: from .convert import convert convert(args.files, args.dest_dir, args.verbose) def tags_f(args: argparse.Namespace) -> None: from .tags import tags names = ( tags( wheel, args.python_tag, args.abi_tag, args.platform_tag, args.build, args.remove, ) for wheel in args.wheel ) for name in names: print(name) def version_f(args: argparse.Namespace) -> None: from .. import __version__ print(f"wheel {__version__}") def parse_build_tag(build_tag: str) -> str: if build_tag and not build_tag[0].isdigit(): raise ArgumentTypeError("build tag must begin with a digit") elif "-" in build_tag: raise ArgumentTypeError("invalid character ('-') in build tag") return build_tag TAGS_HELP = """\ Make a new wheel with given tags. Any tags unspecified will remain the same. Starting the tags with a "+" will append to the existing tags. Starting with a "-" will remove a tag (use --option=-TAG syntax). Multiple tags can be separated by ".". The original file will remain unless --remove is given. The output filename(s) will be displayed on stdout for further processing. """ def parser(): p = argparse.ArgumentParser() s = p.add_subparsers(help="commands") unpack_parser = s.add_parser("unpack", help="Unpack wheel") unpack_parser.add_argument( "--dest", "-d", help="Destination directory", default="." ) unpack_parser.add_argument("wheelfile", help="Wheel file") unpack_parser.set_defaults(func=unpack_f) repack_parser = s.add_parser("pack", help="Repack wheel") repack_parser.add_argument("directory", help="Root directory of the unpacked wheel") repack_parser.add_argument( "--dest-dir", "-d", default=os.path.curdir, help="Directory to store the wheel (default %(default)s)", ) repack_parser.add_argument( "--build-number", help="Build tag to use in the wheel name" ) repack_parser.set_defaults(func=pack_f) convert_parser = s.add_parser("convert", help="Convert egg or wininst to wheel") convert_parser.add_argument("files", nargs="*", help="Files to convert") convert_parser.add_argument( "--dest-dir", "-d", default=os.path.curdir, help="Directory to store wheels (default %(default)s)", ) convert_parser.add_argument("--verbose", "-v", action="store_true") convert_parser.set_defaults(func=convert_f) tags_parser = s.add_parser( "tags", help="Add or replace the tags on a wheel", description=TAGS_HELP ) tags_parser.add_argument("wheel", nargs="*", help="Existing wheel(s) to retag") tags_parser.add_argument( "--remove", action="store_true", help="Remove the original files, keeping only the renamed ones", ) tags_parser.add_argument( "--python-tag", metavar="TAG", help="Specify an interpreter tag(s)" ) tags_parser.add_argument("--abi-tag", metavar="TAG", help="Specify an ABI tag(s)") tags_parser.add_argument( "--platform-tag", metavar="TAG", help="Specify a platform tag(s)" ) tags_parser.add_argument( "--build", type=parse_build_tag, metavar="BUILD", help="Specify a build tag" ) tags_parser.set_defaults(func=tags_f) version_parser = s.add_parser("version", help="Print version and exit") version_parser.set_defaults(func=version_f) help_parser = s.add_parser("help", help="Show this help") help_parser.set_defaults(func=lambda args: p.print_help()) return p def main(): p = parser() args = p.parse_args() if not hasattr(args, "func"): p.print_help() else: try: args.func(args) return 0 except WheelError as e: print(e, file=sys.stderr) return 1 auto_attach.py 0000644 00000001676 15030160121 0007411 0 ustar 00 from uaclient import event_logger, exceptions, messages from uaclient.api.u.pro.attach.auto.full_auto_attach.v1 import ( FullAutoAttachOptions, _full_auto_attach, ) from uaclient.cli import cli_util from uaclient.cli.commands import ProCommand from uaclient.cli.parser import HelpCategory event = event_logger.get_event_logger() @cli_util.assert_root def action_auto_attach(args, *, cfg, **kwargs) -> int: try: _full_auto_attach( FullAutoAttachOptions(), cfg=cfg, mode=event_logger.EventLoggerMode.CLI, ) except exceptions.ConnectivityError: event.info(messages.E_ATTACH_FAILURE.msg) return 1 else: cli_util.post_cli_attach(cfg) return 0 auto_attach_command = ProCommand( "auto-attach", help=messages.CLI_ROOT_AUTO_ATTACH, description=messages.CLI_AUTO_ATTACH_DESC, action=action_auto_attach, help_category=HelpCategory.OTHER, ) parser.py 0000644 00000005220 15030160121 0006376 0 ustar 00 import argparse from collections import OrderedDict from enum import Enum from typing import List, NamedTuple # noqa: F401 from uaclient import messages HelpEntry = NamedTuple( "HelpEntry", [("position", int), ("name", str), ("help_string", str)] ) class HelpCategory(Enum): class _Value: def __init__(self, code: str, msg: str): self.code = code self.msg = msg QUICKSTART = _Value("quickstart", messages.CLI_HELP_HEADER_QUICK_START) SECURITY = _Value("security", messages.CLI_HELP_HEADER_SECURITY) TROUBLESHOOT = _Value( "troubleshoot", messages.CLI_HELP_HEADER_TROUBLESHOOT ) OTHER = _Value("other", messages.CLI_HELP_HEADER_OTHER) FLAGS = _Value("flags", messages.CLI_FLAGS) def __str__(self): return self.value.code @property def header(self): return self.value.msg class ProArgumentParser(argparse.ArgumentParser): help_entries = OrderedDict( [ (HelpCategory.QUICKSTART, []), (HelpCategory.SECURITY, []), (HelpCategory.TROUBLESHOOT, []), (HelpCategory.OTHER, []), (HelpCategory.FLAGS, []), ] ) # type: OrderedDict[HelpCategory, List[HelpEntry]] @classmethod def add_help_entry( cls, category: HelpCategory, name: str, help_string: str, position: int = 0, ): cls.help_entries[category].append( HelpEntry(position=position, name=name, help_string=help_string) ) def __init__(self, *args, use_main_help: bool = True, **kwargs): super().__init__(*args, **kwargs) self.use_main_help = use_main_help def print_help_for_command(self, command: str): args_list = command.split() args_list.append("--help") try: self.parse_args(args_list) # We want help for any specific command, # but without exiting right after except SystemExit: pass def format_help(self): if self.use_main_help: return super().format_help() help_output = self.format_usage() for category, items in self.help_entries.items(): help_output += "\n" help_output += "{}:".format(category.header) help_output += "\n" for item in sorted(items, key=lambda item: item.position): help_output += "\n" help_output += " {:<17}{}".format(item.name, item.help_string) help_output += "\n" if self.epilog: help_output += "\n" help_output += self.epilog help_output += "\n" return help_output disable.py 0000644 00000025167 15030160121 0006521 0 ustar 00 import json import logging import textwrap from typing import Dict, List # noqa: F401 from uaclient import ( config, contract, entitlements, event_logger, exceptions, messages, status, util, ) from uaclient.api import ProgressWrapper from uaclient.api.u.pro.services.dependencies.v1 import ( ServiceWithDependencies, _dependencies, ) from uaclient.api.u.pro.status.enabled_services.v1 import _enabled_services from uaclient.cli import cli_util from uaclient.cli.commands import ProArgument, ProArgumentGroup, ProCommand from uaclient.cli.parser import HelpCategory from uaclient.entitlements.entitlement_status import CanDisableFailure event = event_logger.get_event_logger() LOG = logging.getLogger(util.replace_top_level_logger_name(__name__)) def prompt_for_dependency_handling( cfg: config.UAConfig, service: str, all_dependencies: List[ServiceWithDependencies], enabled_service_names: List[str], called_name: str, service_title: str, ): dependent_services = [] for s in all_dependencies: if s.name == service or s.name not in enabled_service_names: continue for requirement in s.depends_on: if requirement.name == service: dependent_services.append(s.name) for dependent_service in dependent_services: dependent_service_title = entitlements.get_title( cfg, dependent_service ) user_msg = messages.DEPENDENT_SERVICE.format( service_being_disabled=service_title, dependent_service=dependent_service_title, ) if not util.prompt_for_confirmation(msg=user_msg): raise exceptions.DependentServiceStopsDisable( service_being_disabled=service_title, dependent_service=dependent_service_title, ) def perform_disable( entitlement, cfg, *, json_output, assume_yes, update_status=True ): """Perform the disable action on a named entitlement. :param entitlement_name: the name of the entitlement to enable :param cfg: the UAConfig to pass to the entitlement :param json_output: output should be json only @return: True on success, False otherwise """ # Make sure we have the correct variant of the service # This can affect what packages get uninstalled variant = entitlement.enabled_variant if variant is not None: entitlement = variant if json_output: progress = ProgressWrapper() else: progress = ProgressWrapper( cli_util.CLIEnableDisableProgress(assume_yes=assume_yes) ) ret, reason = entitlement.disable(progress) if not ret: event.service_failed(entitlement.name) if reason is not None and isinstance(reason, CanDisableFailure): if reason.message is not None: event.info(reason.message.msg) event.error( error_msg=reason.message.msg, error_code=reason.message.name, service=entitlement.name, ) else: event.service_processed(entitlement.name) if update_status: status.status(cfg=cfg) # Update the status cache return ret @cli_util.verify_json_format_args @cli_util.assert_root @cli_util.assert_attached(cli_util._raise_enable_disable_unattached_error) @cli_util.assert_lock_file("pro disable") def action_disable(args, *, cfg, **kwargs): """Perform the disable action on a list of entitlements. @return: 0 on success, 1 otherwise """ processed_services = [] failed_services = [] errors = [] warnings = [] # type: List[Dict[str, str]] json_response = { "_schema_version": event_logger.JSON_SCHEMA_VERSION, "result": "success", "needs_reboot": False, } json_output = args.format == "json" assume_yes = args.assume_yes # HACK NOTICE: interactive_only_print here will be a no-op "null_print" # function defined above if args.format == "json". We use this function # throughout enable for things that should get printed in the normal # interactive output so that they don't get printed for the json output. interactive_only_print = cli_util.create_interactive_only_print_function( json_output ) if args.purge and assume_yes: raise exceptions.InvalidOptionCombination( option1="--purge", option2="--assume-yes" ) names = getattr(args, "service", []) ( entitlements_found, entitlements_not_found, ) = entitlements.get_valid_entitlement_names(names, cfg) enabled_service_names = [ s.name for s in _enabled_services(cfg).enabled_services ] all_dependencies = _dependencies(cfg).services ret = True for ent_name in entitlements_found: ent = entitlements.entitlement_factory( cfg=cfg, name=ent_name, purge=args.purge, ) variant = ent.enabled_variant if variant is not None: ent = variant if not assume_yes: # this never happens for json output because we assert earlier that # assume_yes must be True for json output try: prompt_for_dependency_handling( cfg, ent.name, all_dependencies, enabled_service_names, called_name=ent_name, service_title=ent.title, ) except exceptions.UbuntuProError as e: LOG.exception(e) interactive_only_print(e.msg) interactive_only_print( messages.ENABLE_FAILED.format(title=ent.title) ) ret = False continue if json_output: progress = ProgressWrapper() else: progress = ProgressWrapper( cli_util.CLIEnableDisableProgress(assume_yes=assume_yes) ) progress.total_steps = ent.calculate_total_disable_steps() try: disable_ret, reason = ent.disable(progress) status.status(cfg=cfg) # Update the status cache if not disable_ret: ret = False failed_services.append(ent_name) if reason is not None and isinstance( reason, CanDisableFailure ): if reason.message is not None: interactive_only_print(reason.message.msg) errors.append( { "type": "service", "service": ent.name, "message": reason.message.msg, "message_code": reason.message.name, } ) else: processed_services.append(ent_name) ent_reboot_required = ent._check_for_reboot() if ent_reboot_required: json_response["needs_reboot"] = True interactive_only_print( messages.ENABLE_REBOOT_REQUIRED_TMPL.format( operation="disable operation" ) ) except exceptions.UbuntuProError as e: ret = False failed_services.append(ent_name) interactive_only_print(e.msg) interactive_only_print( messages.DISABLE_FAILED_TMPL.format(title=ent.title) ) errors.append( { "type": "service", "service": ent.name, "message": e.msg, "message_code": e.msg_code, "additional_info": e.additional_info, } ) if entitlements_not_found: ret = False valid_names = ( "Try " + ", ".join(entitlements.valid_services(cfg=cfg)) + "." ) service_msg = "\n".join( textwrap.wrap( valid_names, width=80, break_long_words=False, break_on_hyphens=False, ) ) err = exceptions.InvalidServiceOpError( operation="disable", invalid_service=", ".join(entitlements_not_found), service_msg=service_msg, ) interactive_only_print(err.msg) errors.append( { "type": "system", "service": None, "message": err.msg, "message_code": err.msg_code, "additional_info": err.additional_info, } ) contract_client = contract.UAContractClient(cfg) contract_client.update_activity_token() if json_output: processed_services.sort() failed_services.sort() json_response["result"] = "success" if ret else "failure" json_response["processed_services"] = processed_services json_response["failed_services"] = failed_services json_response["errors"] = errors json_response["warnings"] = warnings print( json.dumps( json_response, cls=util.DatetimeAwareJSONEncoder, sort_keys=True, ) ) return 0 if ret else 1 disable_command = ProCommand( "disable", help=messages.CLI_ROOT_DISABLE, description=messages.CLI_DISABLE_DESC, action=action_disable, help_category=HelpCategory.OTHER, argument_groups=[ ProArgumentGroup( arguments=[ ProArgument( "service", help=messages.CLI_DISABLE_SERVICE.format( options=", ".join( entitlements.valid_services(cfg=config.UAConfig()) ) ), action="store", nargs="+", ), ProArgument( "--assume-yes", help=messages.CLI_ASSUME_YES.format(command="disable"), action="store_true", ), ProArgument( "--format", help=messages.CLI_FORMAT_DESC.format(default="cli"), action="store", choices=["cli", "json"], default="cli", ), ProArgument( "--purge", help=messages.CLI_PURGE, action="store_true", ), ] ) ], ) status.py 0000644 00000004570 15030160121 0006434 0 ustar 00 import time from uaclient import actions, config, event_logger, messages, status, util from uaclient.cli.commands import ProArgument, ProArgumentGroup, ProCommand from uaclient.cli.parser import HelpCategory event = event_logger.get_event_logger() def action_status(args, *, cfg: config.UAConfig, **kwargs): if not cfg: cfg = config.UAConfig() show_all = args.all if args else False token = args.simulate_with_token if args else None active_value = status.UserFacingConfigStatus.ACTIVE.value status_dict, ret = actions.status( cfg, simulate_with_token=token, show_all=show_all ) config_active = bool(status_dict["execution_status"] == active_value) if args and args.wait and config_active: while status_dict["execution_status"] == active_value: event.info(".", end="") time.sleep(1) status_dict, ret = actions.status( cfg, simulate_with_token=token, show_all=show_all, ) event.info("") event.set_output_content(status_dict) output = status.format_tabular(status_dict, show_all=show_all) event.info(util.handle_unicode_characters(output)) event.process_events() return ret status_command = ProCommand( "status", help=messages.CLI_ROOT_STATUS, description=messages.CLI_STATUS_DESC, action=action_status, preserve_description=True, help_category=HelpCategory.QUICKSTART, help_position=1, argument_groups=[ ProArgumentGroup( arguments=[ ProArgument( "--wait", help=messages.CLI_STATUS_WAIT, action="store_true", ), ProArgument( "--format", help=messages.CLI_FORMAT_DESC.format(default="tabular"), action="store", choices=["tabular", "json", "yaml"], default="tabular", ), ProArgument( "--simulate-with-token", help=messages.CLI_STATUS_SIMULATE_WITH_TOKEN, metavar="TOKEN", action="store", ), ProArgument( "--all", help=messages.CLI_STATUS_ALL, action="store_true" ), ] ) ], ) __pycache__/collect_logs.cpython-310.pyc 0000644 00000002634 15030160121 0014120 0 ustar 00 o �(�f6 � @ s� d dl Z d dlZd dlZd dlmZ d dlmZ d dlmZm Z m Z d dl m Z d dl mZ dZe �ee��Zdd � Ze d ejejee je ed d ejd �gd�gd�ZdS )� N)�messages)� collect_logs)� ProArgument�ProArgumentGroup� ProCommand)� HelpCategory)� replace_top_level_logger_namezpro_logs.tar.gzc K s� | j pt}t�� �L}t||� z t�|d��}|j|dd� W d � n1 s(w Y W n tyL } zt � |� W Y d }~W d � dS d }~ww W d � dS 1 sXw Y dS )Nzw:gzzlogs/)�arcname� r ) �output�PRO_COLLECT_LOGS_FILE�tempfile�TemporaryDirectoryr �tarfile�open�add�PermissionError�LOG�error)�args�cfg�kwargs� output_file� output_dir�results�e� r �;/usr/lib/python3/dist-packages/uaclient/cli/collect_logs.py�action_collect_logs s& �� ��� ��r z collect-logsz--outputz-o)� short_name�help)� arguments)r � description�action� help_category�argument_groups)�loggingr r �uaclientr �uaclient.actionsr �uaclient.cli.commandsr r r �uaclient.cli.parserr � uaclient.utilr r � getLogger�__name__r r �CLI_ROOT_COLLECT_LOGS�CLI_COLLECT_LOGS_DESC� TROUBLESHOOT�CLI_COLLECT_LOGS_OUTPUT�collect_logs_commandr r r r �<module> s6 ���� � __pycache__/system.cpython-310.pyc 0000644 00000002333 15030160121 0012767 0 ustar 00 o �(�f2 � @ s� d dl mZmZ d dlmZ d dlmZ d dlmZ e� � Z dd� Z dd� Z ed ej eje d d �Zed ejeje ejd egd�ZdS )� )� event_logger�messages)�_reboot_required)� ProCommand)� HelpCategoryc K s t |�}t�|j� dS )Nr )r �event�info�reboot_required)�args�cfg�kwargs�result� r �5/usr/lib/python3/dist-packages/uaclient/cli/system.py�action_reboot_required s r c K s ddl m} |� �d� d S )Nr )� get_parser�system)� uaclient.clir �print_help_for_command)r r r r r r r � action_system s r zreboot-requiredT)�help� description�action�preserve_descriptionr � )r r r � help_category� help_position� subcommandsN)�uaclientr r �5uaclient.api.u.pro.security.status.reboot_required.v1r �uaclient.cli.commandsr �uaclient.cli.parserr �get_event_loggerr r r �CLI_SYSTEM_REBOOT_REQUIRED� CLI_SYSTEM_REBOOT_REQUIRED_DESC�reboot_required_subcommand�CLI_ROOT_SYSTEM�CLI_SYSTEM_DESC� QUICKSTART�system_commandr r r r �<module> s. � � __pycache__/detach.cpython-310.pyc 0000644 00000005647 15030160121 0012706 0 ustar 00 o �(�f� � @ s& d dl mZmZmZmZmZmZmZmZ d dl m Z d dl m Z m Z mZ d dlmZ d dlmZ d dlmZmZ d dlmZ e�� Ze je je � � e � d�d e fd d �����Z d ej d e!de!d e fdd�Z"edej#ej$e ej%e e dej&j'dd�dd�e dej(j'dd�dddgdd�gd �gd �Z)d S ) � )�config�daemon� entitlements� event_logger� exceptions�messages�timer�util)�cli_util)� ProArgument�ProArgumentGroup� ProCommand)�perform_disable)� HelpCategory)� machine_token� state_files)�update_motd_messagesz pro detach�returnc K s: t || j| jdkd�}|dkrt�� t�� t�� |S )zXPerform the detach action for this machine. @return: 0 on success, 1 otherwise �json)� assume_yes� json_outputr ) �_detachr �formatr �startr �stop�event�process_events)�args�cfg�kwargs�ret� r! �5/usr/lib/python3/dist-packages/uaclient/cli/detach.py� action_detach s �r# r r r c C s� g }t �| �D ]%}z t j| |d�}W n tjy Y qw |jdd�\}}|r,|�|� q|rHt�t j � t |��� |D ] }t�d� |j�� q<tj|d�sPdS |D ] }t|| ||dd� qRt�| �}|�� t�� t| � t�t j� d S ) aj Detach the machine from the active Ubuntu Pro subscription, :param cfg: a ``config.UAConfig`` instance :param assume_yes: Assume a yes answer to any prompts requested. In this case, it means automatically disable any service during detach. :param json_output: output should be json only @return: 0 on success, 1 otherwise )r �nameT)�ignore_dependent_servicesz {})r � F)r r � update_statusr )r �entitlements_disable_order�entitlement_factoryr �EntitlementNotFoundError� can_disable�appendr �infor �DETACH_WILL_DISABLE� pluralize�lenr r$ r �prompt_for_confirmationr r �get_machine_token_file�deleter �delete_state_filesr �DETACH_SUCCESS) r r r � to_disable�ent_name�entr �_�machine_token_filer! r! r" r ( sD �� � � r �detachz --assume-yes)�command� store_true)�help�actionz--format�cli)�default�storer )r> r? �choicesrA )� arguments)r> � descriptionr? � help_category�argument_groupsN)*�uaclientr r r r r r r r � uaclient.clir �uaclient.cli.commandsr r r �uaclient.cli.disabler �uaclient.cli.parserr �uaclient.filesr r � uaclient.timer.update_messagingr �get_event_loggerr �verify_json_format_args� assert_root�assert_attached�assert_lock_file�intr# �UAConfig�boolr �CLI_ROOT_DETACH�CLI_DETACH_DESC�OTHER�CLI_ASSUME_YESr �CLI_FORMAT_DESC�detach_commandr! r! r! r" �<module> sJ ( 3 � ���� � __pycache__/fix.cpython-310.pyc 0000644 00000050123 15030160121 0012231 0 ustar 00 o �(�f�r � @ s` d dl Z d dlZd dlmZmZmZmZmZmZm Z d dl m Z m Z m Z mZmZ d dlmZmZ d dlmZ d dlmZmZ d dlmZmZ d dlm Z m Z m Z m Z d d l m!Z!m"Z"m#Z#m$Z$m%Z%m&Z&m'Z'm(Z(m)Z)m*Z*m+Z+m,Z,m-Z-m.Z.m/Z/m0Z0m1Z1m2Z2m3Z3m4Z4 d d l5m6Z6 d d l5m7Z8 d d l9m:Z: d d l9m7Z; d d l<m=Z=m>Z> d dl?m@Z@mAZAmBZB d dlCmDZD d dlEmFZF d dlGmHZHmIZImJZJ d dlKmLZL d dlMmNZN d dlOmPZP d dlQmRZRmSZSmTZT d dlUmVZV d dlWmXZX d dlYmZZZ d dl[m\Z\ G dd� d�Z]d e+fd d �Z^d e-fd d!�Z_d"e`d#ead$eLfd%d&�Zbd"e`d#ead'ead$eLfd(d)�Zc dod*ee` d+e`d,edd-edd.ee` d/e`f d0d1�Zed$eLd2e`d/eafd3d4�Zfdpd5d6�Zgd$eLfd7d8�Zhd$eLd/eafd9d:�Zid;ee` d/e`fd<d=�Zjd$eLd#ead/eafd>d?�Zkd$eLd/eafd@dA�Zld$eLdBe`d/eafdCdD�ZmdBe`d$eLd#ead/eafdEdF�Zn Gdqd+e dHe`dIe`fdJdK�ZodLe`fdMdN�ZpdOe]dPe0fdQdR�ZqdOe]dPe1fdSdT�ZrdOe]dPe/fdUdV�ZsdOe]dPe$fdWdX�ZtdOe]dPe%fdYdZ�ZudOe]dPe&fd[d\�ZvdOe]dPe*fd]d^�ZwdOe]dPe(fd_d`�ZxdOe]dPe'fdadb�Zyd e+d#ead$eLd/ee ee f fdcdd�Zzdedf� Z{eBdge j|e j}e{eFj~eAe@d"e jdh�e@die j�djdk�e@dle j�djdk�gdm�gdn�Z�dS )r� N)�Dict�List� NamedTuple�Optional�Set�Tuple�Union)�apt� exceptions�messages�system�util)�attach_with_token�enable_entitlement_by_name)� _initiate)�MagicAttachRevokeOptions�_revoke)�MagicAttachWaitOptions�_wait)�CVE_OR_USN_REGEX� FixStatus�UnfixedPackage�status_message)�ESM_APPS_POCKET�ESM_INFRA_POCKET�STANDARD_UPDATES_POCKET�FixPlanAptUpgradeStep�FixPlanAttachStep�FixPlanEnableStep�FixPlanNoOpAlreadyFixedStep�FixPlanNoOpLivepatchFixStep�FixPlanNoOpStatus�FixPlanNoOpStep� FixPlanResult� FixPlanStep�FixPlanUSNResult�FixPlanWarning�"FixPlanWarningFailUpdatingESMCache�&FixPlanWarningPackageCannotBeInstalled�#FixPlanWarningSecurityIssueNotFixed�NoOpAlreadyFixedData�NoOpLivepatchFixData�USNAdditionalData)�CVEFixPlanOptions)�_plan)�USNFixPlanOptions)�ContractExpiryStatus� _is_attached)� ProArgument�ProArgumentGroup� ProCommand)� action_detach)� HelpCategory)�CLOUD_TYPE_TO_TITLE�PRO_CLOUD_URLS�get_cloud_type)�UAConfig)�PRINT_WRAP_WIDTH)�entitlement_factory)�ApplicabilityStatus�CanEnableFailure�UserFacingStatus)�notices)�Notice)� PRO_HOME_PAGE)�colorize_commandsc @ sj e Zd Zdededee defdd�Zdd� Z dd ee d ed e e fd d�Z dee defdd�Z d S )� FixContext�title�dry_run� affected_pkgs�cfgc C sJ d| _ g | _t� | _tj| _|| _|| _|| _ || _ d| _ d| _ d| _ d S )Nr TF)� pkg_index� unfixed_pkgs�set�installed_pkgsr �SYSTEM_NON_VULNERABLE� fix_statusrE rG rF rH �should_print_pkg_header� warn_package_cannot_be_installed�fixed_by_livepatch)�selfrE rF rG rH � rS �2/usr/lib/python3/dist-packages/uaclient/cli/fix.py�__init__S s zFixContext.__init__c C sR | j r'tj�t| j ��jt| j �d�t| j ��d�}tt j |t ddd�� d S d S )N�, )�count�pkgs� F)�width�subsequent_indent�replace_whitespace) rG r �SECURITY_AFFECTED_PKGS� pluralize�len�format�join�sorted�print�textwrap�fillr; )rR �msgrS rS rT �print_fix_headerf s" �����zFixContext.print_fix_headerN� source_pkgs�status�pocketc C s8 | j rtt||| jt| j�|rt|�nd d�� d S d S )N)�pkg_listri rI �num_pkgs� pocket_source)rO rc �_format_packages_messagerI r_ rG �get_pocket_description)rR rh ri rj rS rS rT �print_pkg_headerw s ���zFixContext.print_pkg_headerrX �unfixed_reasonc C s" |D ] }| j �t||d�� qd S )N)�pkgrq )rJ �appendr )rR rX rq rr rS rS rT �add_unfixed_packages� s ��z FixContext.add_unfixed_packages�N) �__name__� __module__� __qualname__�str�boolr r: rU rg r rp rt rS rS rS rT rD R s( ��� ���� �rD �cvec C s8 dj | j�� | jd�d� | j�� �g}td�|�� d S )N�{issue}: {description}��issue� descriptionz! - https://ubuntu.com/security/{}� )r` rE �upperr rc ra )r{ �linesrS rS rT �print_cve_header� s ��r� �fix_planc C s� | j }dj|j�� |jd�g}|j}t|t�rK|jr5|� t j � |jD ]}|� d�t j j j|d��� q#n|jrK|� t j� |jD ] }|� d| � qAtd�|�� d S )Nr| r} z - {})r{ z - r� )�target_usn_planr` rE r� r �additional_data� isinstancer, �associated_cvesrs r �SECURITY_FOUND_CVES�urls�SECURITY_CVE_PAGE�associated_launchpad_bugs� SECURITY_FOUND_LAUNCHPAD_BUGSrc ra )r� � target_usnr� r� r{ �lp_bugrS rS rT �print_usn_header� s* �� ��� r� �security_issuerF rH c C sz t t| gd�|d�}|jjd j}|r$|jr$tjt� |j p d|j�d��t |jjd � t � t |jjd ||�\}}|S )N)�cves��optionsrH r �unexpected-error�� named_msg)�cve_planr- � cves_datar� �errorrf r �AnonymousUbuntuProErrorr � NamedMessage�coder� rc �execute_fix_plan)r� rF rH r� r� ri �_rS rS rT �fix_cve� s � ��r� � no_relatedc C s� t t| gd�|d�}|jjd jj}|r%|jr%tjt � |j p d|j�d��t |jjd � t dt jj| d� � t|jjd j||�\}}|tjtjfvrO|S |jjd j}|rZ|r\|S t dt jjd�d d � |D ��d � � t dt j � i } |D ]} t d �| j�� t| ||�| | j< t � qzt t j� t|| t jd � d} |D ]=} | | j \} } t| | jt j d � | tj kr�t dt j jdd� � d} | tj kr�| D ]}|j r�t d�|j!|j �� q�d} q�| r�t dt j"j| d� � |S )N)�usnsr� r r� r� r� )�issue_idz - c s s � | ]}|j V qd S ru )rE )�.0�usnrS rS rT � <genexpr>� s � zfix_usn.<locals>.<genexpr>)� related_usnsz- {})�contextF�- � fix operation�� operationTz - {}: {})#�usn_planr/ � usns_datar� r� r� rf r r� r r� r� r� rc � SECURITY_FIXING_REQUESTED_USNr` r� r rM �SYSTEM_NOT_AFFECTED�related_usns_plan�SECURITY_RELATED_USNSra � SECURITY_FIXING_RELATED_USNSrE �SECURITY_USN_SUMMARY�_handle_fix_status_message�FIX_ISSUE_CONTEXT_REQUESTED�FIX_ISSUE_CONTEXT_RELATED� SYSTEM_VULNERABLE_UNTIL_REBOOT�ENABLE_REBOOT_REQUIRED_TMPL�SYSTEM_STILL_VULNERABLErq rr �SECURITY_RELATED_USN_ERROR)r� rF r� rH r� r� �target_usn_statusr� r� �related_usn_status�related_usn_plan�failure_on_related_usnri rJ � unfixed_pkgrS rS rT �fix_usn� s� � ����� ������ � �� ��� �������r� rk ri rI rl rm �returnc C s� | sdS g }g }| D ]}|d7 }|� d�||�� |� |� q tjd�dd�|� d d�t|���tdd �}d �|t||��S ) z;Format the packages and status to an user friendly message.� � z{}/{}z{} {}:�(rV �)rY �rZ r[ z{} {})rs r` rd re ra rb r; r ) rk ri rI rl rm � msg_index�src_pkgs�src_pkg� msg_headerrS rS rT rn 0 s ��rn �tokenc C s\ t tdd|gg�� z t| |dd� W dS tjy- } z t |j� W Y d}~dS d}~ww )ztAttach to an Ubuntu Pro subscription with a given token. :return: True if attach performed without errors. �pro�attachT)r� � allow_enableNF)rc rC r r �UbuntuProErrorrf )rH r� �errrS rS rT �_run_ua_attachL s ��r� c C s> t � \} }| t�� v r ttjjt�| �t�| �d�� dS dS )z:Alert the user when running Pro on cloud with PRO support.)rE �cloud_specific_urlN) r9 r8 �keysrc r �SECURITY_USE_PRO_TMPLr` r7 �get)� cloud_typer� rS rS rT �*_inform_ubuntu_pro_existence_if_applicableZ s ���r� c C s� t tj� t| d�}t dtjj|jd� � t|jd�}zt || d�}W n t j yD } zt tj � t |jd�}t|| d� |�d }~ww t dtj � t| |j�S )N)rH r� )� user_code)� magic_tokenr� )rc r �CLI_MAGIC_ATTACH_INITr �CLI_MAGIC_ATTACH_SIGN_INr` r� r r� r r �MagicAttachTokenError�CLI_MAGIC_ATTACH_FAILEDr r �CLI_MAGIC_ATTACH_PROCESSINGr� �contract_token)rH � initiate_resp� wait_options� wait_resp�e�revoke_optionsrS rS rT �_perform_magic_attachf s. ��� � �� r� c C sh t � ttj� tjtjg d�d�}|dkrdS |dkr t| �S |dkr2ttj� t d�}t | |�S dS ) zZPrompt for attach to a subscription or token. :return: True if attach performed. )�s�a�c�� valid_choicesr� Fr� r� �> T) r� rc r �*SECURITY_UPDATE_NOT_INSTALLED_SUBSCRIPTIONr �prompt_choices�SECURITY_FIX_ATTACH_PROMPTr� �PROMPT_ENTER_TOKEN�inputr� )rH �choicer� rS rS rT �_prompt_for_attach� s � r� rJ c C s4 t | �}tjtj�|�j|d�t| ��d�t dd�S )z�Format the list of unfixed packages into an message. :returns: A string containing the message output for the unfixed packages. rV )rl rX rY r� ) r_ rd re r �SECURITY_PKG_STILL_AFFECTEDr^ r` ra rb r; )rJ �num_pkgs_unfixedrS rS rT � _format_unfixed_packages_msg� s � ��r� c C s4 t | �j}|r|tjjkr|rttj� dS dS dS )zuCheck if the Ubuntu Pro subscription is expired. :returns: True if subscription is expired and not renewed. FT)r1 �contract_statusr0 �EXPIRED�valuerc r �(SECURITY_DRY_RUN_UA_EXPIRED_SUBSCRIPTION)rH rF �contract_expiry_statusrS rS rT � _check_subscription_is_expired� s � r� c C s� ddl }t� ttj� tjtjjt d�ddgd�}|dkr?ttj � t d�}tt dd gg�� t |jd d d �| � t| |�S d S )zdPrompt for attach a new subscription token to the user. :return: True if attach performed. r N)�url�rr� r� r� r� �detachT�cli)� assume_yesr` F)�argparser� rc r �%SECURITY_UPDATE_NOT_INSTALLED_EXPIREDr r� �SECURITY_FIX_RENEW_PROMPTr` rB �PROMPT_EXPIRED_ENTER_TOKENr� rC r5 � Namespacer� )rH r r� r� rS rS rT �_prompt_for_new_token� s � r �servicec C s� t tjj|d�� tjtjj|d�ddgd�}|dkrEt tdd|gg�� t| |d�\}}|sC|durCt |t �rC|j durCt |j j � |S d S ) zMPrompt for enable a pro service. :return: True if enable performed. �r r� r� r� r� �enable�rH �nameNF) rc r �SECURITY_SERVICE_DISABLEDr` r r� �SECURITY_FIX_ENABLE_PROMPTrC r r� r> �messagerf )rH r r� �ret�reasonrS rS rT �_prompt_for_enable� s" ��� r c C s� t || d�}|rN|�� \}}|tjkrdS |�� \}}|tjkrD|r0tdtj j |j d� � dS t ||j �r8dS ttj j |j d�� dS ttjj |j d�� dS )zQ Verify if the Ubuntu Pro subscription has the required service enabled. r Tr� r F)r< �user_facing_statusr? �ACTIVE�applicability_statusr= � APPLICABLErc r �'SECURITY_DRY_RUN_UA_SERVICE_NOT_ENABLEDr` r r � SECURITY_UA_SERVICE_NOT_ENABLED� SECURITY_UA_SERVICE_NOT_ENTITLED)r rH rF �ent� ent_statusr� r rS rS rT �)_handle_subscription_for_required_service� s< ��� �� ���r r� r� r� c C s� | t jkr |rtjj||d�}ntjj|d�}tt�|�� d S | t j kr@|r0tj j||d�}ntj j|d�}tt�|�� d S | t j kr`|rPtj j||d�}ntjj|d�}tt�|�� d S |rktj j||d�}ntjj|d�}tt�|�� d S )N)r~ r� �r~ )r rM r �%SECURITY_ISSUE_RESOLVED_ISSUE_CONTEXTr` �SECURITY_ISSUE_RESOLVEDrc r �handle_unicode_charactersr� �'SECURITY_ISSUE_UNAFFECTED_ISSUE_CONTEXT�SECURITY_ISSUE_UNAFFECTEDr� �)SECURITY_ISSUE_NOT_RESOLVED_ISSUE_CONTEXT�SECURITY_ISSUE_NOT_RESOLVED)ri r� r� rf rS rS rT r� s6 � � ��r� rj c C s. | t krtjS | tkrtjS | tkrtjS | S ru )r r �'SECURITY_UBUNTU_STANDARD_UPDATES_POCKETr �SECURITY_UA_INFRA_POCKETr �SECURITY_UA_APPS_POCKET)rj rS rS rT ro @ s ro � fix_context�stepc C sh | j |jjd|jjd� d| _tjj|jj|jj d�}t d| � | j |jj g|d� d| _ tj| _d S )N�released�rh ri rj F)�package�versionr� �rX rq T)rp �data�related_source_packagesrj rO r �FIX_CANNOT_INSTALL_PACKAGEr` �binary_package�binary_package_versionrc rt �source_packagerP r r� rN )r( r) �warn_msgrS rS rT �)_execute_package_cannot_be_installed_stepK s �� � r6 c C sR | j |jj|jjd� | jt|jj�7 _| j|jjt|jj�d� tj | _ d S )N)rh ri r. ) rp r/ �source_packagesri rI r_ rt r r r� rN �r( r) rS rS rT �&_execute_security_issue_not_fixed_stepd s � � r9 c C s, t �� r ttj� d S tdtj d � d S )Nr� )r �we_are_currently_rootrc r � CLI_FIX_FAIL_UPDATING_ESM_CACHE�(CLI_FIX_FAIL_UPDATING_ESM_CACHE_NON_ROOTr8 rS rS rT �%_execute_fail_updating_esm_cache_stept s r= c C sh | j |jjd|jjd� | jt|jj�7 _|jjs)| js#tt j � t j | _ d S t�� sE| jsEtt j� t j| _ | j|jjt jd� d S ttg d�g d� t|jj� g�� | jrat j | _ d S zt�� tjg d�|jj ddid � W n, ty� } z t|d t|��}t|� t j| _ | j|jj|d� W Y d }~d S d }~ww t j | _ d | _ | j � |jj� d S ) Nr* r+ r. )r �updatez&&)r �install�--only-upgrade�-y)zapt-getr? r@ rA �DEBIAN_FRONTEND�noninteractive)�cmd�override_env_varsrf T) rp r/ r7 rj rI r_ �binary_packagesrP rc r �SECURITY_UPDATE_INSTALLEDr rM rN r r: rF �SECURITY_APT_NON_ROOTr� rt rC rb r �run_apt_update_command�run_apt_command� Exception�getattrry rO rL r> )r( r) r� rf rS rS rT �_execute_apt_upgrade_step} sl � �� ���� � ���� rM c C s� |j jdkrtnt}| j|j jd|d� d| _t| j�j sD| j r(t dt j � nHt| j�sCtj| _| j|j jt jj|j jd�d� d S n,t| j| j d�rp| j rUt t j� nt| j�sptj| _| j|j jt jj|j jd�d� d S tj| _d S ) N� esm-infrar* r+ Fr� r r. )rH rF )r/ �required_servicer r rp r7 rO r1 rH � is_attachedrF rc r � SECURITY_DRY_RUN_UA_NOT_ATTACHEDr� r r� rN rt � SECURITY_UA_SERVICE_REQUIREDr` r� r� r �$SECURITY_UA_SERVICE_WITH_EXPIRED_SUBrM �r( r) rj rS rS rT �_execute_attach_step� sL ��� ��� � �� rU c C st |j jdkrtnt}| j|j jd|d� d| _t|j j| j| j �s7| j |j jt j j |j jd�d� tj| _d S tjS )NrN r* r+ Fr r. )r/ r r r rp r7 rO r rH rF rt r �%SECURITY_UA_SERVICE_NOT_ENABLED_SHORTr` r r� rN rM rT rS rS rT �_execute_enable_step� s0 ������rW c C s* |j jtjjkrttj� tj | _ d S d S ru ) r/ ri r! � NOT_AFFECTEDr� rc r �SECURITY_NO_AFFECTED_PKGSr r� rN r8 rS rS rT � _execute_noop_not_affected_step s �rZ c C s4 t |jt�rttjj| j|jjd�� d| _ d S d S )N)r~ r- T) r� r/ r+ rc r �CVE_FIXED_BY_LIVEPATCHr` rE � patch_versionrQ r8 rS rS rT �%_execute_noop_fixed_by_livepatch_step s �� �r] c C sL t |jt�r$| j|jjd|jjd� ttj� | j t |jj�7 _ d S d S )Nr* r+ ) r� r/ r* rp r7 rj rc r rG rI r_ r8 rS rS rT � _execute_noop_already_fixed_step s � �r^ c C s� g | j �| j�}t| j|| jpg |d�}|�� t|dd� d�D ]t}t|t�r,t ||� t|t �r6t ||� t|t �r@t ||� t|t�rRt||� |jtjkrR nCt|t�rdt||� |jtjkrd n1t|t�rvt||� |jtjkrv n t|t�r�t||� t|t�r�t||� t|t�r�t ||� q t � |j r�t t t t!dd� |j D ����� tj"|_|jtjkr�t#j$|j%d�r�tj&|_t'j(j)dd �}t |� t*j+t,j-dd � |j.s�t/|j| j� |j|j fS ) N)rE rF rG rH c S s | j S ru )�order)�xrS rS rT �<lambda>= s z"execute_fix_plan.<locals>.<lambda>)�keyc S s g | ]}|j �qS rS )rr )r� r� rS rS rT � <listcomp>a s ��z$execute_fix_plan.<locals>.<listcomp>)rL r� r� )0�plan�warningsrD rE �affected_packagesrg rb r� r( r6 r) r9 r' r= r rM rN r rM r rU r rW r" rZ r r] r r^ rc rJ r� �listrK r� r � should_rebootrL r� r r� r` r@ �addrA �ENABLE_REBOOT_REQUIREDrQ r� )r� rF rH � full_planr( r) � reboot_msgrS rS rT r� - s� ��� ������ ��� r� c K sj t �t| j�stj| jd��| jrttj � d| j� � v r(t | j| j|�}|jS t | j| j| j |�}|jS )Nr r{ )�re�matchr r� r � InvalidSecurityIssueIdFormatrF rc r �SECURITY_DRY_RUN_WARNING�lowerr� r� r� � exit_code)�argsrH �kwargsri rS rS rT � action_fix s � ��ru �fix)�helpz --dry-run� store_true)rw �actionz --no-related)� arguments)rw r ry � help_category�argument_groupsru )r� N)r� )�rm rd �typingr r r r r r r �uaclientr r r r r �uaclient.actionsr r �+uaclient.api.u.pro.attach.magic.initiate.v1r �)uaclient.api.u.pro.attach.magic.revoke.v1r r �'uaclient.api.u.pro.attach.magic.wait.v1r r �'uaclient.api.u.pro.security.fix._commonr r r r �/uaclient.api.u.pro.security.fix._common.plan.v1r r r r r r r r r! r" r# r$ r% r& r' r( r) r* r+ r, �+uaclient.api.u.pro.security.fix.cve.plan.v1r- r. r� �+uaclient.api.u.pro.security.fix.usn.plan.v1r/ r� �(uaclient.api.u.pro.status.is_attached.v1r0 r1 �uaclient.cli.commandsr2 r3 r4 �uaclient.cli.detachr5 �uaclient.cli.parserr6 �uaclient.clouds.identityr7 r8 r9 �uaclient.configr: �uaclient.defaultsr; �uaclient.entitlementsr< �(uaclient.entitlements.entitlement_statusr= r> r? �uaclient.filesr@ �uaclient.files.noticesrA �uaclient.messages.urlsrB �uaclient.statusrC rD r� r� ry rz r� r� �intrn r� r� r� r� r� r� r r r r� ro r6 r9 r= rM rU rW rZ r] r^ r� ru � CLI_ROOT_FIX� CLI_FIX_DESC�SECURITY� CLI_FIX_ISSUE�CLI_FIX_DRY_RUN�CLI_FIX_NO_RELATED� fix_commandrS rS rS rT �<module> s: $ X ? ��� �l������ � ��� �,��� �% � �� �� � � �>� �/� �!� �� � � � ��� �R ����� � __pycache__/refresh.cpython-310.pyc 0000644 00000004510 15030160121 0013100 0 ustar 00 o �(�f� � @ s d dl Z d dlmZmZmZmZmZmZ d dlm Z d dl m Z m Z m Z d dlmZ d dlmZmZ e �e�e��Zdejfdd �Ze �� dejfd d ��Zdejfd d �Ze j e � d�dejfdd���Z e dej ej e dej!e e dej"dg d�dd�gd�gd�Z#dS )� N)�apt_news�config�contract� exceptions�messages�util)�cli_util)� ProArgument�ProArgumentGroup� ProCommand)� HelpCategory)� refresh_motd�update_motd_messages�cfgc C sF z|� � W n ty } z t�|� t�� �d }~ww ttj� d S �N) �process_config� RuntimeError�LOG� exceptionr �RefreshConfigFailure�printr �REFRESH_CONFIG_SUCCESS)�argsr �exc� r �6/usr/lib/python3/dist-packages/uaclient/cli/refresh.py�_action_refresh_config s ��r c C s6 zt �|� W n tjy t�� �w ttj� d S r )r �refreshr �ConnectivityError�RefreshContractFailurer r �REFRESH_CONTRACT_SUCCESS)�_argsr r r r �_action_refresh_contract s �r" c C s\ zt |� t� |jrt�|� W n ty& } z t�|� t�� �d }~ww t t j � d S r ) r r r �update_apt_news� Exceptionr r r �RefreshMessagesFailurer r �REFRESH_MESSAGES_SUCCESS)r! r r r r r �_action_refresh_messages s � ��r' z pro refreshc K s^ | j d u s | j dkrt| |� | j d u s| j dkr t| |� | j d u s(| j dkr-t| |� dS )Nr r r r )�targetr r" r' )r r �kwargsr r r �action_refresh. s r* r Tr( �?)r r r )�help�nargs�choices�default)� arguments)r, � description�action�preserve_description� help_category�argument_groups)$�logging�uaclientr r r r r r � uaclient.clir �uaclient.cli.commandsr r r �uaclient.cli.parserr � uaclient.timer.update_messagingr r � getLogger� replace_top_level_logger_name�__name__r �UAConfigr �assert_attachedr" r' � assert_root�assert_lock_filer* �CLI_ROOT_REFRESH�CLI_REFRESH_DESC�OTHER�CLI_REFRESH_TARGET�refresh_commandr r r r �<module> sB ���� � __pycache__/cli_util.cpython-310.pyc 0000644 00000014524 15030160121 0013254 0 ustar 00 o �(�f� � @ s d dl mZ d dlmZ d dlmZmZmZmZm Z m Z m Z m Z m Z mZ d dlmZ d dlmZmZ d dlmZ d dlmZ e �� ZG dd � d ej�Zd d � Z d e fd d�Z d%dd�Z dd� Z dd� Z!d%dd�Z"dd� Z#dd� Z$d ed dfd d �Z%d ed ed!e&d"ee& d df d#d$�Z'dS )&� ��wraps)�Optional) �actions�api�daemon� entitlements� event_logger� exceptions�lock�messages�status�util)� _is_attached)� AptProxyScope�setup_apt_proxy)�UAConfig)� machine_tokenc @ sJ e Zd Zdefdd�Zdededee dee fdd �Zd efd d �Z d S )�CLIEnableDisableProgress� assume_yesc C s | | _ || _d S �N)�is_interactiver )�selfr � r �7/usr/lib/python3/dist-packages/uaclient/cli/cli_util.py�__init__ s z!CLIEnableDisableProgress.__init__� total_steps� done_steps�previous_step_message�current_step_messagec C s |d ur t |� d S d S r )�print)r r r r r r r r �progress s �z!CLIEnableDisableProgress.progress�eventc C s: |dkr t |� d S |dkrt�|| j�st�� �d S d S )N�info�message_operation)r r �handle_message_operationsr r �PromptDeniedError)r r" �payloadr r r � _on_event( s �z"CLIEnableDisableProgress._on_eventN) �__name__� __module__� __qualname__�boolr �intr �strr! r( r r r r r s ��� � r c O s d S r r ��args�kwargsr r r � _null_print0 s r2 � json_outputc C s | rt S tS r )r2 r )r3 r r r �&create_interactive_only_print_function4 s r4 Nc � � fdd�}|S )z1Decorator asserting exclusive access to lock filec � t � �� �fdd��}|S )Nc sF t j�dd�� � |d| i|��}W d � |S 1 s w Y |S )N� )� lock_holder� sleep_time�cfg)r � RetryLock)r: r0 r1 �retval)�fr8 r r �new_f? s ��z0assert_lock_file.<locals>.wrapper.<locals>.new_fr �r= r> �r8 �r= r �wrapper> s z!assert_lock_file.<locals>.wrapperr )r8 rB r r@ r �assert_lock_file; s rC c � t � �� fdd��}|S )z Decorator asserting root userc s t �� st�� �� | i |��S r )r �we_are_currently_rootr �NonRootUserErrorr/ rA r r r> M s zassert_root.<locals>.new_fr r? r rA r � assert_rootJ s rG c rD )z>Decorator to verify if correct params are used for json formatc sH | s � | g|�R i |��S | j dkr| jst�� �� | g|�R i |��S )N�json)�formatr r � CLIJSONFormatRequireAssumeYes)�cmd_argsr0 r1 rA r r r> Z s z&verify_json_format_args.<locals>.new_fr r? r rA r �verify_json_format_argsW s rL c r5 )z�Decorator asserting attached config. :param msg_function: Optional function to generate a custom message if raising an UnattachedError c r6 )Nc sR t |�js �rt| dd�}t| dd�}�|||d� nt�� �� | fd|i|��S )N�command� �service)rM � service_namesr: r: )r � is_attached�getattrr �UnattachedError)r0 r: r1 rM rP )r= �raise_custom_error_functionr r r> n s �z/assert_attached.<locals>.wrapper.<locals>.new_fr r? �rT rA r rB m s z assert_attached.<locals>.wrapperr )rT rB r rU r �assert_attachedg s rV c rD )z&Decorator asserting unattached config.c s< t |�jrt�� }tj|j�dd�d��� | fd|i|��S )N�namerN )� account_namer: )r rQ r �get_machine_token_filer �AlreadyAttachedError�account�get)r0 r: r1 �machine_token_filerA r r r> � s �z"assert_not_attached.<locals>.new_fr r? r rA r �assert_not_attached� s r^ c C sf t j||d�\}}|r |r tjd�|�| d�|�dd��|r(tjd�|�| d��tj| d�|�dd��)z�Raises a custom error for enable/disable commands when unattached. Takes into consideration if the services exist or not, and notify the user accordingly.)�namesr: z, rN )� valid_service� operation�invalid_service� service_msg)r` ra )ra rb rc )r �get_valid_entitlement_namesr � UnattachedMixedServicesError�join� UnattachedValidServicesError� UnattachedInvalidServicesError)rM rP r: �entitlements_found�entitlements_not_foundr r r �&_raise_enable_disable_unattached_error� s( �� ��rk r: �returnc C s| t �| �}|j}|rt�tjj|d�� nt�tj� t � � t � | � t � | �\}}t �|�}t�t�|�� t�� d S )N)� contract_name)r rY rm r" r# r �ATTACH_SUCCESS_TMPLrI � ATTACH_SUCCESS_NO_CONTRACT_NAMEr �stop�cleanupr r �format_tabularr �handle_unicode_characters�process_events)r: r] rm � status_dict�_ret�outputr r r �post_cli_attach� s � rx �scope�set_key� set_valuec C sR |t jkr | j}| j}n |t jkr| j}| j}d|v r |}n|}t|||d� dS )zS Handles setting part the apt proxies - global and uaclient scoped proxies �https)� http_proxy� https_proxy� proxy_scopeN)r �GLOBAL�global_apt_http_proxy�global_apt_https_proxy�UACLIENT�ua_apt_http_proxy�ua_apt_https_proxyr )r: ry rz r{ r} r~ r r r �configure_apt_proxy� s �r� r )(� functoolsr �typingr �uaclientr r r r r r r r r r �(uaclient.api.u.pro.status.is_attached.v1r � uaclient.aptr r �uaclient.configr �uaclient.filesr �get_event_loggerr" �AbstractProgressr r2 r, r4 rC rG rL rV r^ rk rx r. r� r r r r �<module> s: 0 ����� __pycache__/parser.cpython-310.pyc 0000644 00000006273 15030160121 0012746 0 ustar 00 o �(�f� � @ s~ d dl Z d dlmZ d dlmZ d dlmZmZ d dlm Z edde fde fd e fg�Z G d d � d e�Z G d d � d e j�ZdS )� N)� OrderedDict)�Enum)�List� NamedTuple)�messages� HelpEntry�position�name� help_stringc @ sj e Zd ZG dd� d�Zedej�Zedej�Zedej �Z edej �Z edej �Zdd � Zed d � �Zd S ) � HelpCategoryc @ s e Zd Zdedefdd�ZdS )zHelpCategory._Value�code�msgc C s || _ || _d S �N)r r )�selfr r � r �5/usr/lib/python3/dist-packages/uaclient/cli/parser.py�__init__ s z HelpCategory._Value.__init__N)�__name__� __module__� __qualname__�strr r r r r �_Value s r � quickstart�security� troubleshoot�other�flagsc C � | j jS r )�valuer �r r r r �__str__ s zHelpCategory.__str__c C r r )r r r r r r �header s zHelpCategory.headerN)r r r r r �CLI_HELP_HEADER_QUICK_START� QUICKSTART�CLI_HELP_HEADER_SECURITY�SECURITY� CLI_HELP_HEADER_TROUBLESHOOT� TROUBLESHOOT�CLI_HELP_HEADER_OTHER�OTHER� CLI_FLAGS�FLAGSr �propertyr! r r r r r s � r c s� e Zd Zeejg fejg fejg fejg fej g fg�Z e ddede de de fdd��Zdd �d ef� fd d �Zd e fdd�Z� fdd�Z� ZS )�ProArgumentParserr �categoryr r r c C s | j | �t|||d�� d S )N)r r r )� help_entries�appendr )�clsr. r r r r r r �add_help_entry/ s �z ProArgumentParser.add_help_entryT)� use_main_helpr3 c s t � j|i |�� || _d S r )�superr r3 )r r3 �args�kwargs�� __class__r r r ; s zProArgumentParser.__init__�commandc C s8 |� � }|�d� z| �|� W d S ty Y d S w )Nz--help)�splitr0 � parse_args� SystemExit)r r9 � args_listr r r �print_help_for_command? s �z(ProArgumentParser.print_help_for_commandc s� | j rt� �� S | �� }| j�� D ]1\}}|d7 }|d�|j�7 }|d7 }t|dd� d�D ]}|d7 }|d�|j |j �7 }q-|d7 }q| j rS|d7 }|| j 7 }|d7 }|S )N� z{}:c S s | j S r )r )�itemr r r �<lambda>R s z/ProArgumentParser.format_help.<locals>.<lambda>)�keyz {:<17}{}) r3 r4 � format_help� format_usager/ �items�formatr! �sortedr r �epilog)r � help_outputr. rE r@ r7 r r rC I s z ProArgumentParser.format_help)r )r r r r r r# r% r' r) r+ r/ � classmethodr �intr2 �boolr r> rC � __classcell__r r r7 r r- $ s. �� ���� � r- )�argparse� collectionsr �enumr �typingr r �uaclientr rK r r r �ArgumentParserr- r r r r �<module> s � __pycache__/help.cpython-310.pyc 0000644 00000002470 15030160121 0012375 0 ustar 00 o �(�f� � @ s� d dl Z d dlmZmZmZmZ d dlmZmZm Z dd� Z e dej ej e eedej jd�eje�� d ��d �d d d �edejjdd�d g d�dd�edejdd�gd�gd�ZdS )� N)�config� entitlements�messages�status)� ProArgument�ProArgumentGroup� ProCommandc K s� | j }|sddlm} |� �� dS |st�� }t�||�}| jdkr,t t � |�� dS |� � D ]\}}t d�|� � |�� q0dS )Nr )� get_parser�jsonz{}: {} )�service� uaclient.clir � print_helpr �UAConfigr �help�format�printr �dumps�items�title)�args�cfg�kwargsr r � help_response�key�value� r �3/usr/lib/python3/dist-packages/uaclient/cli/help.py� action_help s �r r r z, )r )�options�store�?)r �action�nargsz--format�tabular)�default)r# r �yaml)r r! �choicesr$ z--all� store_true)r r! )� arguments)r � descriptionr! �argument_groups)r �uaclientr r r r �uaclient.cli.commandsr r r r � CLI_ROOT_HELP� CLI_HELP_DESC�CLI_HELP_SERVICEr �join�valid_servicesr �CLI_FORMAT_DESC� CLI_HELP_ALL� help_commandr r r r �<module> sD ��� ����� � __pycache__/enable.cpython-310.pyc 0000644 00000023435 15030160121 0012677 0 ustar 00 o �(�fD � @ s� d dl Z d dlZd dlmZmZmZmZmZ d dlm Z m Z m Z m Z m Z mZmZmZmZmZ d dlmZmZ d dlmZmZmZ d dlmZm Z d dl m Z d dl m Z d d l!m"Z"m#Z#m$Z$ d d l%m&Z& e�'e�(e)��Z*ed d e+fd e+fdeee,ef fg�Z- dBde j.de+dee j/ fdd�Z0de j.de,dee dee de,de,de,fdd�Z1d e+d ee,ef d ee, d ee, d eee,ef d!eee,ef d e+fd"d#�Z2de j.d$e,de,de+d%e+d e+d&eee, dee dee d'e-fd(d)�Z3e j4e j5e �6e j7�d'e8fd*d+����Z9e$d,ej:ej;e9e&j<d-e#e"dej=j>d.�?e j@e �.� d/��d0�d1d2d3�e"d4ejAj>d,d5�d6d7�e"d8ejBd6d7�e"d9ejCd6d7�e"d:ejDj>d;d<�d1d;d=gd;d>�e"d?ejEd1d7�gd@�gdA�ZFdS )C� N)�Any�Dict�List� NamedTuple�Optional) �api�config�contract� entitlements� event_logger� exceptions�lock�messages�status�util)�ServiceWithDependencies� _dependencies)� EnableOptions� EnableResult�_enable)�EnabledService�_enabled_services)� _is_attached)�cli_util)� ProArgument�ProArgumentGroup� ProCommand)� HelpCategory�_EnableOneServiceResult�success� needs_reboot�error�cfg� access_only�progress_objectc C s� t �|�}tj| d||d�}d}d}z tjdd�� |j|d�\}}W d � n1 s,w Y W n tyD } zt�� |�d}~ww |s^|durT|j durT|j } nt j } t j d| d��tdgg dg d �S ) a Landscape gets special treatment because it currently not supported by our enable API. This function is a temporary workaround until we have a proper API for enabling landscape, which will happen after Landscape is fully integrated with the contracts backend. � landscape)� called_namer# � extra_argsFNz cli.enable._enable_landscape)� lock_holder)�progress)�service�reason)�enabled�disabled�reboot_requiredr )r �ProgressWrapperr �LandscapeEntitlementr � RetryLock�enable� Exception�clear_lock_file_if_present�messager �GENERIC_UNKNOWN_ISSUEr �EntitlementNotEnabledErrorr ) r" r# r' r$ r) r% r � fail_reason�er+ � r: �5/usr/lib/python3/dist-packages/uaclient/cli/enable.py�_enable_landscape/ s> �� ����� �r< r* �all_dependencies�enabled_servicesr& �variant� service_titlec s� g }g }dd� |D �� t �fdd�|D �d �} | d ur/� fdd�| jD �}� fdd�| jD �}|D ]'} tj| jdd�} t�| | �} tj j || d �} | sQtj | d �sXt j || d ��q1|D ] }t�| |�}tjj ||d �} tj | d �sxt j||d ��q[t ��fd d�|D �d �}|d ur��d ur�t�| ���}t�| �|j�}tj| jdd�} tj j ||d �} | s�tj | d �s�t j ||d ��d S d S d S ) Nc S s g | ]}|j �qS r: ��name��.0�sr: r: r; � <listcomp>f s z2prompt_for_dependency_handling.<locals>.<listcomp>c 3 s � | ] }|j � kr|V qd S �NrA rC )r* r: r; � <genexpr>i s � z1prompt_for_dependency_handling.<locals>.<genexpr>c s g | ] }|j � v r|j �qS r: rA rC ��enabled_service_namesr: r; rF l � �c s g | ] }|j � vr|j �qS r: rA rC rI r: r; rF q rK z features.block_disable_on_enable)r � path_to_value)�service_being_enabled�incompatible_service)�msg)rM �required_servicec 3 s. � | ]}|j � kr|jr|j�kr|V qd S rG �rB �variant_enabled� variant_namerC )r* r? r: r; rH � s � � ��)�next�incompatible_with� depends_onr �is_config_value_truer" r � get_titler �INCOMPATIBLE_SERVICE�format�prompt_for_confirmationr � IncompatibleServiceStopsEnable�REQUIRED_SERVICE�RequiredServiceStopsEnablerS )r" r* r= r> r& r? r@ �incompatible_services�required_services� dependenciesrN �cfg_block_disable_on_enable�incompatible_service_title�user_msgrP �required_service_titlerR �to_be_enabled_title�enabled_variant_titler: )rJ r* r? r; � prompt_for_dependency_handling[ s� � � ������� � �� �� ������ rh � json_output� json_response�processed_services�failed_services�errors�warningsc C sb | r/|� � |� � |rdnd|d< ||d< ||d< ||d< ||d< ttj|tjdd �� d S d S ) Nr �failure�resultrk rl rm rn T)�cls� sort_keys)�sort�print�json�dumpsr �DatetimeAwareJSONEncoder)ri rj rk rl rm rn r r: r: r; �_print_json_output� s ���rx �ent_name� assume_yesr' �returnc s� t �|�} tj| |�||d�} | j� | j} t� �fdd�|D �d �} | d urItjj | d�} | | j � | tj j | d�� t ddd|| j | jd�d�S |s�z t | � |||�| d � W n- tjy� } z t�|� | |j � | tj j | d�� t ddd d�W Y d }~S d }~ww z_|r�d }nt j|d �}� d kr�t| |||d �}n tt|�|d �| |d�}tj| d� |r�| tjj | d�� n | tjj | d�� |j}|r�| tjj dd�� |jD ]}| |� q�t d|d d�W S tj�y$ } z1|j d }|d }|d }|d }| |� | tj j | d�� t ddd||||d�d�W Y d }~S d }~w tj�yU } z$| |j � | tj j | d�� t ddd||j |j |j d�d�W Y d }~S d }~ww )N)r? r# r' c 3 s2 � | ]}|j � kr�r|jr|j�kr|V qd S rG rQ rC �� real_namer? r: r; rH � s � �� ��z&_enable_one_service.<locals>.<genexpr>)�titleFr* )�typer* r5 � message_code)r r r! )r& r? r@ )rz r% )r' r$ )r* r? r# )r$ �r" �install)� operationTr+ �coder~ �additional_info�r r* r5 r� r� ) r �&create_interactive_only_print_functionr �entitlement_factoryrB r~ rT r �ALREADY_ENABLEDrZ rO � ENABLE_FAILEDr rh r �UbuntuProError�LOG� exception�CLIEnableDisableProgressr< r r r �ACCESS_ENABLED_TMPL� ENABLED_TMPLr. �ENABLE_REBOOT_REQUIRED_TMPLr7 r� �msg_code)r" ry r? r# rz ri r' r> r= �interactive_only_print�ent� ent_title�already_enabledrO r9 r) � enable_resultr r5 r+ �err_code�err_msg�err_infor: r| r; �_enable_one_service� s �� � � �� � ���� ��� � ��� � ��� ����r� c K s( g }g }g }g }t jdd�}| jdk}t�|�} t| dd�} | j} | j} | r/| r/tj ddd��| t j � zt � |� W n! tjtjfy\ tjd d d � |�d t jjt jjd �� Y nw t|�js�t�� } | | j� |�d | j| jd �� t||||||dd� dS t| dg �}t� ||�\}}t |�j }t |�j }d }t�!||�D ]3}t"||| | | ||�#d�||� }|j$r�|�|� |j%r�d |d< q�d}|�|� |j&dur�|�|j&� q�|r�d}||7 }tj'||d�}| |j� |�d d|j|j|j(d�� t �)|�}|�*� t|||||||d� |�rdS dS )z^Perform the enable action on a named entitlement. @return: 0 on success, 1 otherwise F)�_schema_versionr ru r? � � --access-only� --variant)�option1�option2zFailed to refresh contractT)�exc_info�system)r r5 r� )r � r* r' r Nr� r� r )+r �JSON_SCHEMA_VERSIONrZ r r� �getattrr# rz r �InvalidOptionCombinationr �REFRESH_CONTRACT_ENABLEr �refresh�ConnectivityErrorr� r� �warning�append�E_REFRESH_CONTRACT_FAILURErO rB r � is_attached_and_contract_valid�ContractExpiredErrorr� rx r �get_valid_entitlement_namesr r> r �services� order_entitlements_for_enablingr� �getr r r! �*create_enable_entitlements_not_found_errorr� �UAContractClient�update_activity_token)�argsr" �kwargsrk rl rm rn rj ri r� r? r# rz � expired_err�names�entitlements_found�entitlements_not_foundr> r= �retry rp �err�contract_clientr: r: r; � action_enableu s� � � � ��� ��� � �� � �� �� � r� r2 � z, r� )�options�store�+)�help�action�nargsz --assume-yes)�command� store_true)r� r� r� z--betaz--format�cli)�defaultru )r� r� �choicesr� r� )� arguments)r� � descriptionr� � help_category� help_position�argument_groupsrG )Gru �logging�typingr r r r r �uaclientr r r r r r r r r r �+uaclient.api.u.pro.services.dependencies.v1r r �%uaclient.api.u.pro.services.enable.v1r r r �-uaclient.api.u.pro.status.enabled_services.v1r r �(uaclient.api.u.pro.status.is_attached.v1r � uaclient.clir �uaclient.cli.commandsr r r �uaclient.cli.parserr � getLogger� replace_top_level_logger_name�__name__r� �bool�strr �UAConfig�AbstractProgressr< rh rx r� �verify_json_format_args� assert_root�assert_attached�&_raise_enable_disable_unattached_error�intr� �CLI_ROOT_ENABLE�CLI_ENABLE_DESC� QUICKSTART�CLI_ENABLE_SERVICErZ �join�valid_services�CLI_ASSUME_YES�CLI_ENABLE_ACCESS_ONLY�CLI_ENABLE_BETA�CLI_FORMAT_DESC�CLI_ENABLE_VARIANT�enable_commandr: r: r: r; �<module> s 0 ����� �,������ �\� ����� � ������ �� � � # ��� ��� ����� � __pycache__/attach.cpython-310.pyc 0000644 00000007662 15030160121 0012721 0 ustar 00 o �(�f� � @ sh d dl Z d dlZd dlmZmZmZmZmZmZ d dl m Z d dl m Z m Z d dlmZmZ d dlmZ d dlmZmZmZ d dlmZ d d lmZmZ d d l m Z m Z d d l m Z d d l!m"Z" e�#� Z$d d� Z%ej&ej'e�(d�dd� ���Z)edej*ej+e)dej,deedej-dd�edej.ddd�ed ej/e �0d �d �ed ej1j2d d!�d"d d#gd d$�gd%�gd&�Z3dS )'� N)�actions�contract� event_logger� exceptions�messages�secret_manager)� _initiate)�MagicAttachRevokeOptions�_revoke)�MagicAttachWaitOptions�_wait)�cli_util)� ProArgument�ProArgumentGroup� ProCommand)� HelpCategory)�AttachActionsConfigFile�IncorrectTypeError)�*create_enable_entitlements_not_found_error�get_valid_entitlement_names)�CanEnableFailure)� safe_loadc K s� | j dkr tjd| j d��t�tj� t|d�}t�dtjj |j d� � t |j d�}zt ||d�}W n! tj yT } zt�tj� t|j d�}t||d� |�d }~ww t�dtj � |jS ) N�json�--format)�param�value��cfg� )� user_code)� magic_token)�optionsr )�formatr �MagicAttachInvalidParam�event�infor �CLI_MAGIC_ATTACH_INITr �CLI_MAGIC_ATTACH_SIGN_INr r �tokenr �MagicAttachTokenError�CLI_MAGIC_ATTACH_FAILEDr r �CLI_MAGIC_ATTACH_PROCESSING�contract_token)�argsr �kwargs� initiate_resp� wait_options� wait_resp�e�revoke_options� r4 �5/usr/lib/python3/dist-packages/uaclient/cli/attach.py� _magic_attach# s8 � ��� � �� r6 z pro attachc K s� | j r | jr t�� �| j s| jst| |d�}d }n6| j r(| j }tj�|� d }n'z t� t | j��}W n t yH } z tj | jj |jd��d }~ww |j }|j}| joU|d u }z tj|||d� W n tjyl t�� �w d}|d ur�| jr�t||�\} } | D ]5} t�|| �\} } | s�d}| d ur�t| t�r�| jd ur�t�| jj� tj | jj| jj | d� qt� | � q| r�t | |d�}tj|jt j d� tj |j|j!d� d}t"�#|�}|�$� t%�&|� |S ) Nr )� config_name�error)r( � allow_enabler � )� error_msg� error_code�service)� file_type)r; r< )'r( � attach_configr �CLIAttachTokenArgXORConfigr6 r �secrets� add_secretr � from_dictr r � AttachInvalidConfigFileError�name�msg�enable_services� auto_enabler �attach_with_token�ConnectivityError� AttachErrorr �enable_entitlement_by_name� isinstancer �messager$ r% r8 �service_processedr �sys�stderr�msg_coder �UAContractClient�update_activity_tokenr �post_cli_attach)r- r r. r( �enable_services_overrider? r2 r9 �ret�found� not_foundrE �ent_ret�reasonr8 �contract_clientr4 r4 r5 � action_attachD sv � ������ �� � r] �attachT� r( �?)�help�nargsz--no-auto-enable� store_falserH )ra �action�destz--attach-config�r)ra �typer �cli)�default�storer )ra rd �choicesri )� arguments)ra � descriptionrd �preserve_description� help_category� help_position�argument_groups)4�argparserP �uaclientr r r r r r �+uaclient.api.u.pro.attach.magic.initiate.v1r �)uaclient.api.u.pro.attach.magic.revoke.v1r r �'uaclient.api.u.pro.attach.magic.wait.v1r r � uaclient.clir �uaclient.cli.commandsr r r �uaclient.cli.parserr �uaclient.data_typesr r �uaclient.entitlementsr r �(uaclient.entitlements.entitlement_statusr � uaclient.yamlr �get_event_loggerr$ r6 �assert_not_attached� assert_root�assert_lock_filer] �CLI_ROOT_ATTACH�CLI_ATTACH_DESC� QUICKSTART�CLI_ATTACH_TOKEN�CLI_ATTACH_NO_AUTO_ENABLE�CLI_ATTACH_ATTACH_CONFIG�FileType�CLI_FORMAT_DESCr" �attach_commandr4 r4 r4 r5 �<module> sj !D��� ���� � __pycache__/api.cpython-310.pyc 0000644 00000004036 15030160121 0012216 0 ustar 00 o �(�f} � @ s� d dl Z d dlZd dlmZ d dlmZmZ d dlmZm Z d dl m Z d dl m Z d dlmZmZmZ d dlmZ G d d � d e �Zd d � Zed e je jeejeede jdd�ede jdd�ede jdg dd�ede j ddd�gd �gd �Z dS ) � N)� OrderedDict)�Any�Optional)� exceptions�messages)�AbstractProgress)�call_api)� ProArgument�ProArgumentGroup� ProCommand)� HelpCategoryc @ s. e Zd Zdededee dee fdd�ZdS )�CLIAPIProgress� total_steps� done_steps�previous_step_message�current_step_messagec C s8 t � }||d< ||d< ||d< ||d< tt�|�� d S )Nr r r r )r �print�json�dumps)�selfr r r r �d� r �2/usr/lib/python3/dist-packages/uaclient/cli/api.py�progress s zCLIAPIProgress.progressN)�__name__� __module__� __qualname__�intr �strr r r r r r s ����r c K s� | j r | jr t�� �| jr | jdkr tj�� s tj�� | _| jr$t � }nd }t | j | j | j||�}t |� � � |jdkr>dS dS )N�-�successr � )�options�datar �CLIAPIOptionsXORData�sys�stdin�isatty�read� show_progressr r � endpoint_pathr �to_json�result)�args�cfg�kwargsr r, r r r � action_api s � r0 �apir* �endpoint)�help�metavarz--show-progress� store_true)r3 �actionz--argsr" �*)r3 �dest�default�nargsz--datar# � )r3 r8 r9 )� arguments)r3 � descriptionr6 � help_category�argument_groups) r r% � collectionsr �typingr r �uaclientr r � uaclient.apir �uaclient.api.apir �uaclient.cli.commandsr r r �uaclient.cli.parserr r r0 � CLI_ROOT_API� CLI_API_DESC�OTHER�CLI_API_ENDPOINT�CLI_API_SHOW_PROGRESS� CLI_API_ARGS� CLI_API_DATA� api_commandr r r r �<module> sZ ������� � __pycache__/config.cpython-310.pyc 0000644 00000013713 15030160121 0012714 0 ustar 00 o �(�f�' � @ s� d dl mZmZmZmZmZmZmZ d dlm Z d dl m Z d dl m Z mZmZ d dlmZ d dlmZ d dlmZ d dlmZmZ d d lmZmZ e� � Z d d � Z d d � Z e j dd� �Z!e j dd� �Z"edej#ej#e ee dej$dd�gd�gd�Z%edej&ej&e!ee dej'j(d�)ej*�d�d �gd�gd�Z+ed ej,ej,e"ee dej-j(d�)ej*�d�dd �gd�gd�Z.ed ej/ej0e ej1e%e+e.gd �Z2d!S )"� )�apt_news�config� entitlements� event_logger� exceptions�http�messages)� AptProxyScope)�cli_util)� ProArgument�ProArgumentGroup� ProCommand)� HelpCategory)�ApplicationStatus)� state_files)�configure_livepatch_proxy�unconfigure_livepatch_proxy)�configure_snap_proxy�unconfigure_snap_proxyc K s ddl m} |� �d� dS )Nr �� get_parserr )� uaclient.clir �print_help_for_command)�args�cfg�kwargsr � r �5/usr/lib/python3/dist-packages/uaclient/cli/config.py� action_config s r c K s� | j r)| j tjvrtjd�| j �d�tj�d��tdj| j t|| j d�d�� dS t t dd � tjD ��d �}d | d }tjD ]}t|j|t||d�d�� q@|j sV|j rc|j s\|jrettj� dS dS dS ) z�Perform the 'config show' action optionally limit output to a single key :return: 0 on success :raise UbuntuProError: on invalid keys z'{}'�, ��arg�choicesz {key} {value}N��key�valuer c S s g | ]}t |��qS r )�len)�.0�xr r r � <listcomp>4 s z&action_config_show.<locals>.<listcomp>� z{key: <z } {value})r$ r �UA_CONFIGURABLE_KEYSr �InvalidArgChoice�format�join�print�getattr�str�max�global_apt_http_proxy�global_apt_https_proxy�ua_apt_http_proxy�ua_apt_https_proxyr � CLI_CONFIG_GLOBAL_XOR_UA_PROXY)r r r � col_width�row_tmplr$ r r r �action_config_show! s. ��� ���r: c K s ddl m} |� }z | j�d�\}}W n ty' |�d� tjd| jd��w |tj vr=|�d� tj dd� tj �d ��|� � sL|�d� tj |d ��|d v r�|�d �d }|d kr_tj}ntj}t�|||� ||i}td i |�� tj�|�} | �� \} } | tjkr�td i |�� n�||jv r�|�d �d }|d kr�tj}ntj }t�|||� t |j p�|j �} | r�t t!j"j#ddd�� t$�%|t&j'||� d|_ d|_ n�||j(|j) v �r2d|v r�dnd }|d kr�tj}ntj }||j(v �rt t!j*j#d�#|�d�#|�d�� d| }t�|||� t |j+�p|j,�} | �r"t t!j"j#ddd�� t$�%|t&j-||� d|_+d|_,nI|dv �rbzt.|�}|dk �rHtd�#|���W n1 t�ya |�d� t d� tj/||d��w |d k�r{|�0� d k}|�rvt1�2|� nt3j4�5� t6|||� dS ) zMPerform the 'config set' action. @return: 0 on success, 1 otherwise r r �=z config setz <key>=<value>)�expected�actual�<key>r r )r! �� http_proxy� https_proxy�_r � zpro scoped aptz global apt)� current_proxy�previous_proxyN�https� apt_{}_proxy�global_apt_{}_proxy��old�new�global_)�update_messaging_timer�metering_timerzInvalid interval for {}� r# r �truer )7r r �key_value_pair�split� ValueErrorr r �GenericInvalidFormatr r+ r, r. �strip�EmptyConfigValuer � PROXY_VALIDATION_SNAP_HTTP_URL� PROXY_VALIDATION_SNAP_HTTPS_URL�validate_proxyr r � livepatch�LivepatchEntitlement�application_statusr �ENABLEDr �ua_scoped_proxy_options� PROXY_VALIDATION_APT_HTTP_URL� PROXY_VALIDATION_APT_HTTPS_URL�boolr3 r4 r/ r �WARNING_APT_PROXY_OVERWRITEr- r �configure_apt_proxyr �UACLIENT�&deprecated_global_scoped_proxy_options�global_scoped_proxy_options�WARNING_CONFIG_FIELD_RENAMEr5 r6 �GLOBAL�int�InvalidPosIntConfigValue�lowerr �update_apt_newsr �apt_news_contents_file�delete�setattr) r r r r �parser�set_key� set_value� protocol_type� validate_url� entitlement�livepatch_statusrB � unset_currentr r r �action_config_set@ s� �� � � ��� ��� ���� � � �� rx c K s6 ddl m} | jtjvr |� }|�d� tjdd�tj�d��| jdv rH| j� d�d }t |d � t j � |�}|�� \}}|tjkrGt|d � nJ| j|jv rYt�|tj| jd � n9| j|j|j v r�| j|jv r�d | jv rod nd }t�tj j d � |�d� |�d�� d| j | _t�|tj | jd � t || jd � dS )zOPerform the 'config unset' action. @return: 0 on success, 1 otherwise r r z config unsetr> r r r? rB )rs NrF r rG rH rI rL ) r r r$ r r+ r r r, r. rR r r rZ r[ r\ r r] r r^ r rc r rd re rf �event�infor rg r- rh ro ) r r r r rp rs ru rv rB r r r �action_config_unset� sH � � ��� �� r{ �showr$ �?)�help�nargs)� arguments)r~ � description�action�argument_groups�setrQ r )�options)r~ �unset)r~ �metavarr )r~ r� r� � help_category� subcommandsN)3�uaclientr r r r r r r � uaclient.aptr r r �uaclient.cli.commandsr r r �uaclient.cli.parserr �(uaclient.entitlements.entitlement_statusr �uaclient.filesr �uaclient.livepatchr r � uaclient.snapr r �get_event_loggerry r r: � assert_rootrx r{ �CLI_CONFIG_SHOW_DESC�CLI_CONFIG_SHOW_KEY�show_subcommand�CLI_CONFIG_SET_DESC�CLI_CONFIG_SET_KEY_VALUEr- r. r+ �set_subcommand�CLI_CONFIG_UNSET_DESC�CLI_CONFIG_UNSET_KEY�unset_subcommand�CLI_ROOT_CONFIG�CLI_CONFIG_DESC�OTHER�config_commandr r r r �<module> s� $ { ,����� ������ ������ � __pycache__/commands.cpython-310.pyc 0000644 00000007416 15030160121 0013253 0 ustar 00 o �(�f� � @ st d dl Z d dlmZmZmZmZ d dlmZ d dlm Z G dd� d�Z G dd� d�Z G d d � d �Z G d d � d �Z dS ) � N)�Callable�Iterable�Optional�Union)�messages)� HelpCategoryc @ sD e Zd Z d dededee fdd�Zdeejej f fdd �Z dS ) � ProArgumentN� long_name�help� short_namec K s( |d u r|fn||f| _ || _|| _d S �N)�namesr �additional_args)�selfr r r �kwargs� r �7/usr/lib/python3/dist-packages/uaclient/cli/commands.py�__init__ s � zProArgument.__init__�targetc C s |j | jd| ji| j�� d S )Nr )� add_argumentr r r )r r r r r �register s �� �zProArgument.registerr ) �__name__� __module__� __qualname__�strr r r �argparse�ArgumentParser�_ArgumentGroupr r r r r r s ��� � �r c @ s( e Zd Z ddedee fdd�ZdS ) �!ProArgumentMutuallyExclusiveGroupFr �required� argumentsc C s || _ || _d S r )r r )r r r r r r r s z*ProArgumentMutuallyExclusiveGroup.__init__N)Fr )r r r �boolr r r r r r r r s ���r c @ sP e Zd Z d dee dee dee dee fdd�Zd e j fd d �Z dS ) �ProArgumentGroupNr �title� descriptionr �mutually_exclusive_groupsc C s || _ || _|| _|| _d S r )r# r$ r r% )r r# r$ r r% r r r r ) s zProArgumentGroup.__init__�parserc C sd |}| j r |�| j | j�}| jD ]}|�|� q| jD ]}|j|jd�}|jD ]}|�|� q'qd S )N)r )r# �add_argument_groupr$ r r r% � add_mutually_exclusive_groupr )r r& r �argument�group� new_groupr r r r 7 s � � ��zProArgumentGroup.register)NNr r ) r r r r r r r r r r r r r r r r r" ( s" ����� �r" c @ sh e Zd Zdd� dddddfdeded ed ed ed ee d ed dee de fdd�Z de j fdd�ZdS )� ProCommandc O s d S r r )�argsr r r r �<lambda>O s zProCommand.<lambda>Fr Nr �namer r$ �action�preserve_description�argument_groups� subcommands� help_category� help_positionc C s: || _ || _|| _|| _|| _|| _|| _|| _| | _d S r ) r/ r r$ r0 r1 r2 r3 r4 r5 ) r r/ r r$ r0 r1 r2 r3 r4 r5 r r r r J s zProCommand.__init__� subparsersc C s� |j | j| j| jd�| _| jrtj| j_| j r%| jj | j | j| j| j d� | j D ]}|� | j� q(| jj| jd� | jrP| jjtjdd�}| jD ] }|� |� qHd S d S )N)r r$ )�categoryr/ � help_string�position)r0 �command)r# �dest)� add_parserr/ r r$ r& r1 r �RawDescriptionHelpFormatter�formatter_classr4 �add_help_entryr5 r2 r � set_defaultsr0 r3 �add_subparsersr �CLI_AVAILABLE_COMMANDS)r r6 �argument_groupr: r r r r ` s4 � � � �zProCommand.register)r r r r r r! r r" r r �intr r �_SubParsersActionr r r r r r, I s6 �������� � �r, )r �typingr r r r �uaclientr �uaclient.cli.parserr r r r" r, r r r r �<module> s ! __pycache__/auto_attach.cpython-310.pyc 0000644 00000002144 15030160121 0013737 0 ustar 00 o �(�f� � @ s� d dl mZmZmZ d dlmZmZ d dlmZ d dl m Z d dl m Z e� � Zejdefdd��Ze d ejejee jd �Zd S ) � )� event_logger� exceptions�messages)�FullAutoAttachOptions�_full_auto_attach)�cli_util)� ProCommand)� HelpCategory�returnc K sL z t t� |tjjd� W n tjy t�t j j � Y dS w t � |� dS )N)�cfg�mode� r )r r r �EventLoggerMode�CLIr �ConnectivityError�event�infor �E_ATTACH_FAILURE�msgr �post_cli_attach)�argsr �kwargs� r �:/usr/lib/python3/dist-packages/uaclient/cli/auto_attach.py�action_auto_attach s �� r z auto-attach)�help� description�action� help_categoryN)�uaclientr r r �2uaclient.api.u.pro.attach.auto.full_auto_attach.v1r r � uaclient.clir �uaclient.cli.commandsr �uaclient.cli.parserr �get_event_loggerr � assert_root�intr �CLI_ROOT_AUTO_ATTACH�CLI_AUTO_ATTACH_DESC�OTHER�auto_attach_commandr r r r �<module> s � __pycache__/security_status.cpython-310.pyc 0000644 00000003475 15030160121 0014725 0 ustar 00 o �(�f, � @ s� d dl Z d dlmZmZmZ d dlmZmZmZm Z d dl m Z d dl m Z dd� Ze dejejd ee jd eed ejjd d �dd d�geedejdd�edejdd�edejdd�edejdd�gd�gd�gd�ZdS )� N)�messages�security_status�util)� ProArgument�ProArgumentGroup�!ProArgumentMutuallyExclusiveGroup� ProCommand)� HelpCategory)� safe_dumpc K s� | j dkr2| jrt�� dS | jrt�� dS | jr!t�|� dS | jr+t� |� dS t�|� dS | j dkrGt t j t� |�dtjd�� dS t tt� |�dd�� dS )N�text�jsonT)� sort_keys�clsF)�default_flow_styler )�format� thirdpartyr �list_third_party_packages� unavailable�list_unavailable_packages� esm_infra�list_esm_infra_packages�esm_apps�list_esm_apps_packages�printr �dumps�security_status_dictr �DatetimeAwareJSONEncoderr )�args�cfg�kwargs� r �>/usr/lib/python3/dist-packages/uaclient/cli/security_status.py�action_security_status s> �� � � ������r" zsecurity-statusT� z--formatr )�default)r �yamlr )�help�choicesr$ z --thirdparty� store_true)r& �actionz --unavailablez --esm-infraz --esm-apps)� arguments)r* �mutually_exclusive_groups)r& � description�preserve_descriptionr) � help_category� help_position�argument_groups)r �uaclientr r r �uaclient.cli.commandsr r r r �uaclient.cli.parserr � uaclient.yamlr r" �CLI_ROOT_SECURITY_STATUS� CLI_SS_DESC� QUICKSTART�CLI_FORMAT_DESCr �CLI_SS_THIRDPARTY�CLI_SS_UNAVAILABLE�CLI_SS_ESM_INFRA�CLI_SS_ESM_APPS�security_status_commandr r r r! �<module> sb �� ��������� � __pycache__/disable.cpython-310.pyc 0000644 00000014360 15030160121 0013051 0 ustar 00 o �(�fw* � @ s� d dl Z d dlZd dlZd dlmZmZ d dlmZmZm Z m Z m Z m Z m Z mZ d dlmZ d dlmZmZ d dlmZ d dlmZ d dlmZmZmZ d d l m Z d d l m Z e � � Z!e�"e�#e$��Z%d ej&d e'd ee dee' de'de'f dd�Z(dd�dd�Z)ej*ej+e�,ej-�e�.d�dd� ����Z/ede j0e j1e/e j2eed e j3j4d�5e j6e�&� d ��d �d d d �ed!e j7j4dd"�d#d$�ed%e j8j4d&d'�d d&d(gd&d)�ed*e j9d#d$�gd+�gd,�Z:dS )-� N)�Dict�List)�config�contract� entitlements� event_logger� exceptions�messages�status�util)�ProgressWrapper)�ServiceWithDependencies� _dependencies)�_enabled_services)�cli_util)� ProArgument�ProArgumentGroup� ProCommand)� HelpCategory)�CanDisableFailure�cfg�service�all_dependencies�enabled_service_names� called_name� service_titlec C s� g }|D ] }|j |ks|j |vrq|jD ] }|j |kr!|�|j � qq|D ] } t�| | �} tjj|| d�} tj | d�sBt j || d��q%d S )N)�service_being_disabled�dependent_service)�msg) �name� depends_on�appendr � get_titler �DEPENDENT_SERVICE�formatr �prompt_for_confirmationr � DependentServiceStopsDisable) r r r r r r �dependent_services�s� requirementr �dependent_service_title�user_msg� r, �6/usr/lib/python3/dist-packages/uaclient/cli/disable.py� prompt_for_dependency_handling s0 ���� ���r. T)� update_statusc C s� | j }|dur |} |rt� }nttj|d��}| �|�\}}|sIt�| j� |durHt|t �rH|j durHt� |j j � tj |j j |j j| jd� nt�| j� |rWtj|d� |S )a Perform the disable action on a named entitlement. :param entitlement_name: the name of the entitlement to enable :param cfg: the UAConfig to pass to the entitlement :param json_output: output should be json only @return: True on success, False otherwise N�� assume_yes)� error_msg� error_coder �r )�enabled_variantr r �CLIEnableDisableProgress�disable�event�service_failedr � isinstancer �message�infor �error�service_processedr ) � entitlementr � json_outputr1 r/ �variant�progress�ret�reasonr, r, r- �perform_disable> s0 � �� rE z pro disablec K sf g }g }g }g }t jddd�}| jdk}| j} t�|�} | jr(| r(tjddd��t | dg �} t � | |�\} } d d � t |�j D �}t|�j}d }| D ]�}t j||| jd �}|j}|d ur]|}| s�zt||j||||jd� W n) tjy� } z t�|� | |j� | tjj|jd�� d}W Y d }~qId }~ww |r�t� }nttj | d��}|� � |_ zT|� |�\}}t j |d� |s�d}|�!|� |d ur�t"|t#�r�|j$d ur�| |j$j� |�!d|j|j$j|j$jd�� n|�!|� |�%� }|r�d |d< | tj&jdd�� W qI tj�y7 } z+d}|�!|� | |j� | tj'j|jd�� |�!d|j|j|j(|j)d�� W Y d }~qId }~ww | �rtd}dd�*t j+|d�� d }d�*t,j-|dddd ��}tj.d d�*| �|d �}| |j� |�!d d |j|j(|j)d�� t/�0|�} | �1� |�r�|�2� |�2� |�r�dnd |d!< ||d"< ||d#< ||d$< ||d%< t3t4j5|t6j7d d&�� |�r�d'S d(S ))zbPerform the disable action on a list of entitlements. @return: 0 on success, 1 otherwise �successF)�_schema_version�result� needs_reboot�json�--purge� --assume-yes)�option1�option2r c S s g | ]}|j �qS r, )r )�.0r( r, r, r- � <listcomp>� s �z"action_disable.<locals>.<listcomp>T)r r �purgeN)r r )�titler0 r4 )�typer r; � message_coderI zdisable operation)� operation)rS r r; rT �additional_infozTry �, �.� �P )�width�break_long_words�break_on_hyphensr7 )rU �invalid_service� service_msg�system�failurerH �processed_services�failed_services�errors�warnings)�cls� sort_keysr � )8r �JSON_SCHEMA_VERSIONr$ r1 r �&create_interactive_only_print_functionrQ r �InvalidOptionCombination�getattrr �get_valid_entitlement_namesr �enabled_servicesr �services�entitlement_factoryr5 r. r rR �UbuntuProError�LOG� exceptionr r � ENABLE_FAILEDr r6 � calculate_total_disable_steps� total_stepsr7 r r! r: r r; �_check_for_reboot�ENABLE_REBOOT_REQUIRED_TMPL�DISABLE_FAILED_TMPL�msg_coderV �join�valid_services�textwrap�wrap�InvalidServiceOpErrorr �UAContractClient�update_activity_token�sort�printrJ �dumpsr �DatetimeAwareJSONEncoder) �argsr �kwargsrb rc rd re � json_responser@ r1 �interactive_only_print�names�entitlements_found�entitlements_not_foundr r rC �ent_name�entrA �erB � disable_retrD �ent_reboot_required� valid_namesr_ �err�contract_clientr, r, r- �action_disablel s0 � � � �� � � � �� � � ��� ��� ��������� �� ��r� r7 rW r4 )�options�store�+)�help�action�nargsrL )�command� store_true)r� r� z--format�cli)�defaultrJ )r� r� �choicesr� rK )� arguments)r� � descriptionr� � help_category�argument_groups);rJ �loggingr} �typingr r �uaclientr r r r r r r r � uaclient.apir �+uaclient.api.u.pro.services.dependencies.v1r r �-uaclient.api.u.pro.status.enabled_services.v1r � uaclient.clir �uaclient.cli.commandsr r r �uaclient.cli.parserr �(uaclient.entitlements.entitlement_statusr �get_event_loggerr8 � getLogger� replace_top_level_logger_name�__name__rr �UAConfig�strr. rE �verify_json_format_args� assert_root�assert_attached�&_raise_enable_disable_unattached_error�assert_lock_filer� �CLI_ROOT_DISABLE�CLI_DISABLE_DESC�OTHER�CLI_DISABLE_SERVICEr$ r{ r| �CLI_ASSUME_YES�CLI_FORMAT_DESC� CLI_PURGE�disable_commandr, r, r, r- �<module> s� ( ����� � �. 6��� � ����� � __pycache__/status.cpython-310.pyc 0000644 00000003510 15030160121 0012764 0 ustar 00 o �(�fx � @ s� d dl Z d dlmZmZmZmZmZmZ d dlm Z m Z m Z d dl m Z e�� Zdejfdd�Ze dejejed e jd e e d ejd d �e dejjdd�dg d�dd�e dejddd�e dejd d �gd�gd�ZdS )� N)�actions�config� event_logger�messages�status�util)� ProArgument�ProArgumentGroup� ProCommand)� HelpCategory�cfgc K s� |st �� }| r | jnd}| r| jnd }tjjj}tj|||d�\}}t |d |k�}| rY| j rY|rY|d |krTt j ddd� t �d� tj|||d�\}}|d |ks8t � d� t �|� tj||d�} t � t�| �� t �� |S ) NF)�simulate_with_token�show_all�execution_status�.� )�end� )r )r �UAConfig�allr r �UserFacingConfigStatus�ACTIVE�valuer �bool�wait�event�info�time�sleep�set_output_content�format_tabularr �handle_unicode_characters�process_events) �argsr �kwargsr �token� active_value� status_dict�ret� config_active�output� r+ �5/usr/lib/python3/dist-packages/uaclient/cli/status.py� action_status s2 � � � r- r Tr z--wait� store_true)�help�actionz--format�tabular)�default�store)r1 �json�yaml)r/ r0 �choicesr2 z--simulate-with-token�TOKEN)r/ �metavarr0 z--all)� arguments)r/ � descriptionr0 �preserve_description� help_category� help_position�argument_groups)r �uaclientr r r r r r �uaclient.cli.commandsr r r �uaclient.cli.parserr �get_event_loggerr r r- �CLI_ROOT_STATUS�CLI_STATUS_DESC� QUICKSTART�CLI_STATUS_WAIT�CLI_FORMAT_DESC�format� CLI_STATUS_SIMULATE_WITH_TOKEN�CLI_STATUS_ALL�status_commandr+ r+ r+ r, �<module> sP � ������ � enable.py 0000644 00000042020 15030160121 0006327 0 ustar 00 import json import logging from typing import Any, Dict, List, NamedTuple, Optional from uaclient import ( api, config, contract, entitlements, event_logger, exceptions, lock, messages, status, util, ) from uaclient.api.u.pro.services.dependencies.v1 import ( ServiceWithDependencies, _dependencies, ) from uaclient.api.u.pro.services.enable.v1 import ( EnableOptions, EnableResult, _enable, ) from uaclient.api.u.pro.status.enabled_services.v1 import ( EnabledService, _enabled_services, ) from uaclient.api.u.pro.status.is_attached.v1 import _is_attached from uaclient.cli import cli_util from uaclient.cli.commands import ProArgument, ProArgumentGroup, ProCommand from uaclient.cli.parser import HelpCategory LOG = logging.getLogger(util.replace_top_level_logger_name(__name__)) _EnableOneServiceResult = NamedTuple( "_EnableOneServiceResult", [ ("success", bool), ("needs_reboot", bool), ("error", Optional[Dict[str, Any]]), ], ) def _enable_landscape( cfg: config.UAConfig, access_only: bool, extra_args, progress_object: Optional[api.AbstractProgress] = None, ): """ Landscape gets special treatment because it currently not supported by our enable API. This function is a temporary workaround until we have a proper API for enabling landscape, which will happen after Landscape is fully integrated with the contracts backend. """ progress = api.ProgressWrapper(progress_object) landscape = entitlements.LandscapeEntitlement( cfg, called_name="landscape", access_only=access_only, extra_args=extra_args, ) success = False fail_reason = None try: with lock.RetryLock( lock_holder="cli.enable._enable_landscape", ): success, fail_reason = landscape.enable(progress=progress) except Exception as e: lock.clear_lock_file_if_present() raise e if not success: if fail_reason is not None and fail_reason.message is not None: reason = fail_reason.message else: reason = messages.GENERIC_UNKNOWN_ISSUE raise exceptions.EntitlementNotEnabledError( service="landscape", reason=reason ) return EnableResult( enabled=["landscape"], disabled=[], reboot_required=False, messages=[] ) def prompt_for_dependency_handling( cfg: config.UAConfig, service: str, all_dependencies: List[ServiceWithDependencies], enabled_services: List[EnabledService], called_name: str, variant: str, service_title: str, ): incompatible_services = [] required_services = [] enabled_service_names = [s.name for s in enabled_services] dependencies = next( (s for s in all_dependencies if s.name == service), None ) if dependencies is not None: incompatible_services = [ s.name for s in dependencies.incompatible_with if s.name in enabled_service_names ] required_services = [ s.name for s in dependencies.depends_on if s.name not in enabled_service_names ] for incompatible_service in incompatible_services: cfg_block_disable_on_enable = util.is_config_value_true( config=cfg.cfg, path_to_value="features.block_disable_on_enable", ) incompatible_service_title = entitlements.get_title( cfg, incompatible_service ) user_msg = messages.INCOMPATIBLE_SERVICE.format( service_being_enabled=service_title, incompatible_service=incompatible_service_title, ) if cfg_block_disable_on_enable or not util.prompt_for_confirmation( msg=user_msg ): raise exceptions.IncompatibleServiceStopsEnable( service_being_enabled=service_title, incompatible_service=incompatible_service_title, ) for required_service in required_services: required_service_title = entitlements.get_title(cfg, required_service) user_msg = messages.REQUIRED_SERVICE.format( service_being_enabled=service_title, required_service=required_service_title, ) if not util.prompt_for_confirmation(msg=user_msg): raise exceptions.RequiredServiceStopsEnable( service_being_enabled=service_title, required_service=required_service_title, ) variant_enabled = next( ( s for s in enabled_services if s.name == service and s.variant_enabled and s.variant_name != variant ), None, ) if variant_enabled is not None and variant is not None: to_be_enabled_title = entitlements.get_title(cfg, service, variant) enabled_variant_title = entitlements.get_title( cfg, service, variant_enabled.variant_name ) cfg_block_disable_on_enable = util.is_config_value_true( config=cfg.cfg, path_to_value="features.block_disable_on_enable", ) user_msg = messages.INCOMPATIBLE_SERVICE.format( service_being_enabled=to_be_enabled_title, incompatible_service=enabled_variant_title, ) if cfg_block_disable_on_enable or not util.prompt_for_confirmation( msg=user_msg ): raise exceptions.IncompatibleServiceStopsEnable( service_being_enabled=to_be_enabled_title, incompatible_service=enabled_variant_title, ) def _print_json_output( json_output: bool, json_response: Dict[str, Any], processed_services: List[str], failed_services: List[str], errors: List[Dict[str, Any]], warnings: List[Dict[str, Any]], success: bool, ): if json_output: processed_services.sort() failed_services.sort() json_response["result"] = "success" if success else "failure" json_response["processed_services"] = processed_services json_response["failed_services"] = failed_services json_response["errors"] = errors json_response["warnings"] = warnings print( json.dumps( json_response, cls=util.DatetimeAwareJSONEncoder, sort_keys=True, ) ) def _enable_one_service( cfg: config.UAConfig, ent_name: str, variant: str, access_only: bool, assume_yes: bool, json_output: bool, extra_args: Optional[List[str]], enabled_services: List[EnabledService], all_dependencies: List[ServiceWithDependencies], ) -> _EnableOneServiceResult: interactive_only_print = cli_util.create_interactive_only_print_function( json_output ) ent = entitlements.entitlement_factory( cfg, ent_name, variant=variant, access_only=access_only, extra_args=extra_args, ) real_name = ent.name ent_title = ent.title already_enabled = next( ( s for s in enabled_services if s.name == real_name and ( not variant or (s.variant_enabled and s.variant_name == variant) ) ), None, ) if already_enabled is not None: msg = messages.ALREADY_ENABLED.format(title=ent_title) interactive_only_print(msg.msg) interactive_only_print(messages.ENABLE_FAILED.format(title=ent_title)) return _EnableOneServiceResult( success=False, needs_reboot=False, error={ "type": "service", "service": ent_name, "message": msg.msg, "message_code": msg.name, }, ) if not assume_yes: # this never happens for json output because we assert earlier that # assume_yes must be True for json output try: prompt_for_dependency_handling( cfg, real_name, all_dependencies, enabled_services, called_name=ent_name, variant=variant, service_title=ent_title, ) except exceptions.UbuntuProError as e: LOG.exception(e) interactive_only_print(e.msg) interactive_only_print( messages.ENABLE_FAILED.format(title=ent_title) ) return _EnableOneServiceResult( success=False, needs_reboot=False, error=None, ) try: if json_output: progress = None else: progress = cli_util.CLIEnableDisableProgress(assume_yes=assume_yes) if real_name == "landscape": enable_result = _enable_landscape( cfg, access_only, extra_args=extra_args, progress_object=progress, ) else: enable_result = _enable( EnableOptions( service=ent_name, variant=variant, access_only=access_only, ), cfg, progress_object=progress, ) status.status(cfg=cfg) # Update the status cache if access_only: interactive_only_print( messages.ACCESS_ENABLED_TMPL.format(title=ent_title) ) else: interactive_only_print( messages.ENABLED_TMPL.format(title=ent_title) ) needs_reboot = enable_result.reboot_required if needs_reboot: interactive_only_print( messages.ENABLE_REBOOT_REQUIRED_TMPL.format( operation="install" ) ) for message in enable_result.messages: interactive_only_print(message) return _EnableOneServiceResult( success=True, needs_reboot=needs_reboot, error=None, ) except exceptions.EntitlementNotEnabledError as e: reason = e.additional_info["reason"] err_code = reason["code"] err_msg = reason["title"] err_info = reason["additional_info"] interactive_only_print(err_msg) interactive_only_print(messages.ENABLE_FAILED.format(title=ent_title)) return _EnableOneServiceResult( success=False, needs_reboot=False, error={ "type": "service", "service": ent_name, "message": err_msg, "message_code": err_code, "additional_info": err_info, }, ) except exceptions.UbuntuProError as e: interactive_only_print(e.msg) interactive_only_print(messages.ENABLE_FAILED.format(title=ent_title)) return _EnableOneServiceResult( success=False, needs_reboot=False, error={ "type": "service", "service": ent_name, "message": e.msg, "message_code": e.msg_code, "additional_info": e.additional_info, }, ) @cli_util.verify_json_format_args @cli_util.assert_root @cli_util.assert_attached(cli_util._raise_enable_disable_unattached_error) def action_enable(args, *, cfg, **kwargs) -> int: """Perform the enable action on a named entitlement. @return: 0 on success, 1 otherwise """ processed_services = [] # type: List[str] failed_services = [] # type: List[str] errors = [] # type: List[Dict[str, Any]] warnings = [] json_response = { "_schema_version": event_logger.JSON_SCHEMA_VERSION, "needs_reboot": False, } json_output = args.format == "json" # HACK NOTICE: interactive_only_print here will be a no-op "null_print" # function defined above if args.format == "json". We use this function # throughout enable for things that should get printed in the normal # interactive output so that they don't get printed for the json output. interactive_only_print = cli_util.create_interactive_only_print_function( json_output ) variant = getattr(args, "variant", "") access_only = args.access_only assume_yes = args.assume_yes if variant and access_only: raise exceptions.InvalidOptionCombination( option1="--access-only", option2="--variant" ) interactive_only_print(messages.REFRESH_CONTRACT_ENABLE) try: contract.refresh(cfg) except (exceptions.ConnectivityError, exceptions.UbuntuProError): # Inability to refresh is not a critical issue during enable LOG.warning("Failed to refresh contract", exc_info=True) warnings.append( { "type": "system", "message": messages.E_REFRESH_CONTRACT_FAILURE.msg, "message_code": messages.E_REFRESH_CONTRACT_FAILURE.name, } ) if not _is_attached(cfg).is_attached_and_contract_valid: expired_err = exceptions.ContractExpiredError() interactive_only_print(expired_err.msg) errors.append( { "type": "system", "message": expired_err.msg, "message_code": expired_err.msg_code, } ) _print_json_output( json_output, json_response, processed_services, failed_services, errors, warnings, success=False, ) return 1 names = getattr(args, "service", []) ( entitlements_found, entitlements_not_found, ) = entitlements.get_valid_entitlement_names(names, cfg) enabled_services = _enabled_services(cfg).enabled_services all_dependencies = _dependencies(cfg).services ret = True for ent_name in entitlements.order_entitlements_for_enabling( cfg, entitlements_found ): result = _enable_one_service( cfg, ent_name, variant, access_only, assume_yes, json_output, kwargs.get("extra_args"), enabled_services, all_dependencies, ) if result.success: processed_services.append(ent_name) if result.needs_reboot: json_response["needs_reboot"] = True else: ret = False failed_services.append(ent_name) if result.error is not None: errors.append(result.error) if entitlements_not_found: ret = False failed_services += entitlements_not_found err = entitlements.create_enable_entitlements_not_found_error( entitlements_not_found, cfg=cfg ) interactive_only_print(err.msg) errors.append( { "type": "system", "service": None, "message": err.msg, "message_code": err.msg_code, "additional_info": err.additional_info, } ) contract_client = contract.UAContractClient(cfg) contract_client.update_activity_token() _print_json_output( json_output, json_response, processed_services, failed_services, errors, warnings, success=ret, ) return 0 if ret else 1 enable_command = ProCommand( "enable", help=messages.CLI_ROOT_ENABLE, description=messages.CLI_ENABLE_DESC, action=action_enable, help_category=HelpCategory.QUICKSTART, help_position=3, argument_groups=[ ProArgumentGroup( arguments=[ ProArgument( "service", help=messages.CLI_ENABLE_SERVICE.format( options=", ".join( entitlements.valid_services(cfg=config.UAConfig()) ) ), action="store", nargs="+", ), ProArgument( "--assume-yes", help=messages.CLI_ASSUME_YES.format(command="enable"), action="store_true", ), ProArgument( "--access-only", help=messages.CLI_ENABLE_ACCESS_ONLY, action="store_true", ), ProArgument( "--beta", help=messages.CLI_ENABLE_BETA, action="store_true", ), ProArgument( "--format", help=messages.CLI_FORMAT_DESC.format(default="cli"), action="store", choices=["cli", "json"], default="cli", ), ProArgument( "--variant", help=messages.CLI_ENABLE_VARIANT, action="store", ), ] ) ], ) refresh.py 0000644 00000004660 15030160121 0006547 0 ustar 00 import logging from uaclient import apt_news, config, contract, exceptions, messages, util from uaclient.cli import cli_util from uaclient.cli.commands import ProArgument, ProArgumentGroup, ProCommand from uaclient.cli.parser import HelpCategory from uaclient.timer.update_messaging import refresh_motd, update_motd_messages LOG = logging.getLogger(util.replace_top_level_logger_name(__name__)) def _action_refresh_config(args, cfg: config.UAConfig): try: cfg.process_config() except RuntimeError as exc: LOG.exception(exc) raise exceptions.RefreshConfigFailure() print(messages.REFRESH_CONFIG_SUCCESS) @cli_util.assert_attached() def _action_refresh_contract(_args, cfg: config.UAConfig): try: contract.refresh(cfg) except exceptions.ConnectivityError: raise exceptions.RefreshContractFailure() print(messages.REFRESH_CONTRACT_SUCCESS) def _action_refresh_messages(_args, cfg: config.UAConfig): # Not performing any exception handling here since both of these # functions should raise UbuntuProError exceptions, which are # covered by the main_error_handler decorator try: update_motd_messages(cfg) refresh_motd() if cfg.apt_news: apt_news.update_apt_news(cfg) except Exception as exc: LOG.exception(exc) raise exceptions.RefreshMessagesFailure() else: print(messages.REFRESH_MESSAGES_SUCCESS) @cli_util.assert_root @cli_util.assert_lock_file("pro refresh") def action_refresh(args, *, cfg: config.UAConfig, **kwargs): if args.target is None or args.target == "config": _action_refresh_config(args, cfg) if args.target is None or args.target == "contract": _action_refresh_contract(args, cfg) if args.target is None or args.target == "messages": _action_refresh_messages(args, cfg) return 0 refresh_command = ProCommand( "refresh", help=messages.CLI_ROOT_REFRESH, description=messages.CLI_REFRESH_DESC, action=action_refresh, preserve_description=True, help_category=HelpCategory.OTHER, argument_groups=[ ProArgumentGroup( arguments=[ ProArgument( "target", help=messages.CLI_REFRESH_TARGET, nargs="?", choices=["contract", "config", "messages"], default=None, ) ] ) ], ) security_status.py 0000644 00000005454 15030160121 0010365 0 ustar 00 import json from uaclient import messages, security_status, util from uaclient.cli.commands import ( ProArgument, ProArgumentGroup, ProArgumentMutuallyExclusiveGroup, ProCommand, ) from uaclient.cli.parser import HelpCategory from uaclient.yaml import safe_dump def action_security_status(args, *, cfg, **kwargs): if args.format == "text": if args.thirdparty: security_status.list_third_party_packages() elif args.unavailable: security_status.list_unavailable_packages() elif args.esm_infra: security_status.list_esm_infra_packages(cfg) elif args.esm_apps: security_status.list_esm_apps_packages(cfg) else: security_status.security_status(cfg) elif args.format == "json": print( json.dumps( security_status.security_status_dict(cfg), sort_keys=True, cls=util.DatetimeAwareJSONEncoder, ) ) else: print( safe_dump( security_status.security_status_dict(cfg), default_flow_style=False, ) ) return 0 security_status_command = ProCommand( "security-status", help=messages.CLI_ROOT_SECURITY_STATUS, description=messages.CLI_SS_DESC, preserve_description=True, action=action_security_status, help_category=HelpCategory.QUICKSTART, help_position=5, argument_groups=[ ProArgumentGroup( arguments=[ ProArgument( "--format", help=messages.CLI_FORMAT_DESC.format(default="text"), choices=("json", "yaml", "text"), default="text", ) ], mutually_exclusive_groups=[ ProArgumentMutuallyExclusiveGroup( arguments=[ ProArgument( "--thirdparty", help=messages.CLI_SS_THIRDPARTY, action="store_true", ), ProArgument( "--unavailable", help=messages.CLI_SS_UNAVAILABLE, action="store_true", ), ProArgument( "--esm-infra", help=messages.CLI_SS_ESM_INFRA, action="store_true", ), ProArgument( "--esm-apps", help=messages.CLI_SS_ESM_APPS, action="store_true", ), ] ) ], ) ], ) collect_logs.py 0000644 00000002466 15030160121 0007564 0 ustar 00 import logging import tarfile import tempfile from uaclient import messages from uaclient.actions import collect_logs from uaclient.cli.commands import ProArgument, ProArgumentGroup, ProCommand from uaclient.cli.parser import HelpCategory from uaclient.util import replace_top_level_logger_name PRO_COLLECT_LOGS_FILE = "pro_logs.tar.gz" LOG = logging.getLogger(replace_top_level_logger_name(__name__)) def action_collect_logs(args, *, cfg, **kwargs): output_file = args.output or PRO_COLLECT_LOGS_FILE with tempfile.TemporaryDirectory() as output_dir: collect_logs(cfg, output_dir) try: with tarfile.open(output_file, "w:gz") as results: results.add(output_dir, arcname="logs/") except PermissionError as e: LOG.error(e) return 1 return 0 collect_logs_command = ProCommand( "collect-logs", help=messages.CLI_ROOT_COLLECT_LOGS, description=messages.CLI_COLLECT_LOGS_DESC, action=action_collect_logs, help_category=HelpCategory.TROUBLESHOOT, argument_groups=[ ProArgumentGroup( arguments=[ ProArgument( "--output", short_name="-o", help=messages.CLI_COLLECT_LOGS_OUTPUT, ) ] ) ], ) system.py 0000644 00000002062 15030160121 0006427 0 ustar 00 from uaclient import event_logger, messages from uaclient.api.u.pro.security.status.reboot_required.v1 import ( _reboot_required, ) from uaclient.cli.commands import ProCommand from uaclient.cli.parser import HelpCategory event = event_logger.get_event_logger() def action_reboot_required(args, *, cfg, **kwargs): result = _reboot_required(cfg) event.info(result.reboot_required) return 0 def action_system(args, *, cfg, **kwargs): # Avoiding a circular import from uaclient.cli import get_parser get_parser().print_help_for_command("system") reboot_required_subcommand = ProCommand( "reboot-required", help=messages.CLI_SYSTEM_REBOOT_REQUIRED, description=messages.CLI_SYSTEM_REBOOT_REQUIRED_DESC, action=action_reboot_required, preserve_description=True, ) system_command = ProCommand( "system", help=messages.CLI_ROOT_SYSTEM, description=messages.CLI_SYSTEM_DESC, action=action_system, help_category=HelpCategory.QUICKSTART, help_position=4, subcommands=[reboot_required_subcommand], ) config.py 0000644 00000023634 15030160121 0006360 0 ustar 00 from uaclient import ( apt_news, config, entitlements, event_logger, exceptions, http, messages, ) from uaclient.apt import AptProxyScope from uaclient.cli import cli_util from uaclient.cli.commands import ProArgument, ProArgumentGroup, ProCommand from uaclient.cli.parser import HelpCategory from uaclient.entitlements.entitlement_status import ApplicationStatus from uaclient.files import state_files from uaclient.livepatch import ( configure_livepatch_proxy, unconfigure_livepatch_proxy, ) from uaclient.snap import configure_snap_proxy, unconfigure_snap_proxy event = event_logger.get_event_logger() def action_config(args, *, cfg, **kwargs): # Avoiding a circular import from uaclient.cli import get_parser get_parser().print_help_for_command("config") return 0 def action_config_show(args, *, cfg, **kwargs): """Perform the 'config show' action optionally limit output to a single key :return: 0 on success :raise UbuntuProError: on invalid keys """ if args.key: # limit reporting config to a single config key if args.key not in config.UA_CONFIGURABLE_KEYS: raise exceptions.InvalidArgChoice( arg="'{}'".format(args.key), choices=", ".join(config.UA_CONFIGURABLE_KEYS), ) print( "{key} {value}".format( key=args.key, value=getattr(cfg, args.key, None) ) ) return 0 col_width = str(max([len(x) for x in config.UA_CONFIGURABLE_KEYS]) + 1) row_tmpl = "{key: <" + col_width + "} {value}" for key in config.UA_CONFIGURABLE_KEYS: print(row_tmpl.format(key=key, value=getattr(cfg, key, None))) if (cfg.global_apt_http_proxy or cfg.global_apt_https_proxy) and ( cfg.ua_apt_http_proxy or cfg.ua_apt_https_proxy ): print(messages.CLI_CONFIG_GLOBAL_XOR_UA_PROXY) @cli_util.assert_root def action_config_set(args, *, cfg, **kwargs): """Perform the 'config set' action. @return: 0 on success, 1 otherwise """ from uaclient.cli import get_parser parser = get_parser() try: set_key, set_value = args.key_value_pair.split("=") except ValueError: parser.print_help_for_command("config set") raise exceptions.GenericInvalidFormat( expected="<key>=<value>", actual=args.key_value_pair ) if set_key not in config.UA_CONFIGURABLE_KEYS: parser.print_help_for_command("config set") raise exceptions.InvalidArgChoice( arg="<key>", choices=", ".join(config.UA_CONFIGURABLE_KEYS) ) if not set_value.strip(): parser.print_help_for_command("config set") raise exceptions.EmptyConfigValue(arg=set_key) if set_key in ("http_proxy", "https_proxy"): protocol_type = set_key.split("_")[0] if protocol_type == "http": validate_url = http.PROXY_VALIDATION_SNAP_HTTP_URL else: validate_url = http.PROXY_VALIDATION_SNAP_HTTPS_URL http.validate_proxy(protocol_type, set_value, validate_url) kwargs = {set_key: set_value} configure_snap_proxy(**kwargs) # Only set livepatch proxy if livepatch is enabled entitlement = entitlements.livepatch.LivepatchEntitlement(cfg) livepatch_status, _ = entitlement.application_status() if livepatch_status == ApplicationStatus.ENABLED: configure_livepatch_proxy(**kwargs) elif set_key in cfg.ua_scoped_proxy_options: protocol_type = set_key.split("_")[2] if protocol_type == "http": validate_url = http.PROXY_VALIDATION_APT_HTTP_URL else: validate_url = http.PROXY_VALIDATION_APT_HTTPS_URL http.validate_proxy(protocol_type, set_value, validate_url) unset_current = bool( cfg.global_apt_http_proxy or cfg.global_apt_https_proxy ) if unset_current: print( messages.WARNING_APT_PROXY_OVERWRITE.format( current_proxy="pro scoped apt", previous_proxy="global apt" ) ) cli_util.configure_apt_proxy( cfg, AptProxyScope.UACLIENT, set_key, set_value ) cfg.global_apt_http_proxy = None cfg.global_apt_https_proxy = None elif set_key in ( cfg.deprecated_global_scoped_proxy_options + cfg.global_scoped_proxy_options ): # setup_apt_proxy is destructive for unprovided values. Source complete # current config values from uaclient.conf before applying set_value. protocol_type = "https" if "https" in set_key else "http" if protocol_type == "http": validate_url = http.PROXY_VALIDATION_APT_HTTP_URL else: validate_url = http.PROXY_VALIDATION_APT_HTTPS_URL if set_key in cfg.deprecated_global_scoped_proxy_options: print( messages.WARNING_CONFIG_FIELD_RENAME.format( old="apt_{}_proxy".format(protocol_type), new="global_apt_{}_proxy".format(protocol_type), ) ) set_key = "global_" + set_key http.validate_proxy(protocol_type, set_value, validate_url) unset_current = bool(cfg.ua_apt_http_proxy or cfg.ua_apt_https_proxy) if unset_current: print( messages.WARNING_APT_PROXY_OVERWRITE.format( current_proxy="global apt", previous_proxy="pro scoped apt" ) ) cli_util.configure_apt_proxy( cfg, AptProxyScope.GLOBAL, set_key, set_value ) cfg.ua_apt_http_proxy = None cfg.ua_apt_https_proxy = None elif set_key in ( "update_messaging_timer", "metering_timer", ): try: set_value = int(set_value) if set_value < 0: raise ValueError("Invalid interval for {}".format(set_key)) except ValueError: parser.print_help_for_command("config set") # More readable in the CLI, without breaking the line in the logs print("") raise exceptions.InvalidPosIntConfigValue( key=set_key, value=set_value ) elif set_key == "apt_news": set_value = set_value.lower() == "true" if set_value: apt_news.update_apt_news(cfg) else: state_files.apt_news_contents_file.delete() setattr(cfg, set_key, set_value) @cli_util.assert_root def action_config_unset(args, *, cfg, **kwargs): """Perform the 'config unset' action. @return: 0 on success, 1 otherwise """ from uaclient.cli import get_parser if args.key not in config.UA_CONFIGURABLE_KEYS: parser = get_parser() parser.print_help_for_command("config unset") raise exceptions.InvalidArgChoice( arg="<key>", choices=", ".join(config.UA_CONFIGURABLE_KEYS) ) if args.key in ("http_proxy", "https_proxy"): protocol_type = args.key.split("_")[0] unconfigure_snap_proxy(protocol_type=protocol_type) # Only unset livepatch proxy if livepatch is enabled entitlement = entitlements.livepatch.LivepatchEntitlement(cfg) livepatch_status, _ = entitlement.application_status() if livepatch_status == ApplicationStatus.ENABLED: unconfigure_livepatch_proxy(protocol_type=protocol_type) elif args.key in cfg.ua_scoped_proxy_options: cli_util.configure_apt_proxy( cfg, AptProxyScope.UACLIENT, args.key, None ) elif args.key in ( cfg.deprecated_global_scoped_proxy_options + cfg.global_scoped_proxy_options ): if args.key in cfg.deprecated_global_scoped_proxy_options: protocol_type = "https" if "https" in args.key else "http" event.info( messages.WARNING_CONFIG_FIELD_RENAME.format( old="apt_{}_proxy".format(protocol_type), new="global_apt_{}_proxy".format(protocol_type), ) ) args.key = "global_" + args.key cli_util.configure_apt_proxy(cfg, AptProxyScope.GLOBAL, args.key, None) setattr(cfg, args.key, None) return 0 show_subcommand = ProCommand( "show", help=messages.CLI_CONFIG_SHOW_DESC, description=messages.CLI_CONFIG_SHOW_DESC, action=action_config_show, argument_groups=[ ProArgumentGroup( arguments=[ ProArgument( "key", help=messages.CLI_CONFIG_SHOW_KEY, nargs="?" ) ] ) ], ) set_subcommand = ProCommand( "set", help=messages.CLI_CONFIG_SET_DESC, description=messages.CLI_CONFIG_SET_DESC, action=action_config_set, argument_groups=[ ProArgumentGroup( arguments=[ ProArgument( "key_value_pair", help=( messages.CLI_CONFIG_SET_KEY_VALUE.format( options=", ".join(config.UA_CONFIGURABLE_KEYS) ) ), ) ] ) ], ) unset_subcommand = ProCommand( "unset", help=messages.CLI_CONFIG_UNSET_DESC, description=messages.CLI_CONFIG_UNSET_DESC, action=action_config_unset, argument_groups=[ ProArgumentGroup( arguments=[ ProArgument( "key", help=( messages.CLI_CONFIG_UNSET_KEY.format( options=", ".join(config.UA_CONFIGURABLE_KEYS) ) ), metavar="key", ) ] ) ], ) config_command = ProCommand( "config", help=messages.CLI_ROOT_CONFIG, description=messages.CLI_CONFIG_DESC, action=action_config, help_category=HelpCategory.OTHER, subcommands=[show_subcommand, set_subcommand, unset_subcommand], ) api.py 0000644 00000004575 15030160121 0005667 0 ustar 00 import json import sys from collections import OrderedDict from typing import Any, Optional # noqa: F401 from uaclient import exceptions, messages from uaclient.api import AbstractProgress from uaclient.api.api import call_api from uaclient.cli.commands import ProArgument, ProArgumentGroup, ProCommand from uaclient.cli.parser import HelpCategory class CLIAPIProgress(AbstractProgress): def progress( self, *, total_steps: int, done_steps: int, previous_step_message: Optional[str], current_step_message: Optional[str] ): d = OrderedDict() # type: OrderedDict[str, Any] d["total_steps"] = total_steps d["done_steps"] = done_steps d["previous_step_message"] = previous_step_message d["current_step_message"] = current_step_message print(json.dumps(d)) def action_api(args, *, cfg, **kwargs): if args.options and args.data: raise exceptions.CLIAPIOptionsXORData() if args.data and args.data == "-": if not sys.stdin.isatty(): args.data = sys.stdin.read() if args.show_progress: progress = CLIAPIProgress() else: progress = None result = call_api( args.endpoint_path, args.options, args.data, cfg, progress ) print(result.to_json()) return 0 if result.result == "success" else 1 api_command = ProCommand( "api", help=messages.CLI_ROOT_API, description=messages.CLI_API_DESC, action=action_api, help_category=HelpCategory.OTHER, argument_groups=[ ProArgumentGroup( arguments=[ ProArgument( "endpoint_path", help=messages.CLI_API_ENDPOINT, metavar="endpoint", ), ProArgument( "--show-progress", help=messages.CLI_API_SHOW_PROGRESS, action="store_true", ), ProArgument( "--args", help=messages.CLI_API_ARGS, dest="options", default=[], nargs="*", ), ProArgument( "--data", help=messages.CLI_API_DATA, dest="data", default="", ), ] ) ], ) detach.py 0000644 00000006610 15030160121 0006336 0 ustar 00 from uaclient import ( config, daemon, entitlements, event_logger, exceptions, messages, timer, util, ) from uaclient.cli import cli_util from uaclient.cli.commands import ProArgument, ProArgumentGroup, ProCommand from uaclient.cli.disable import perform_disable from uaclient.cli.parser import HelpCategory from uaclient.files import machine_token, state_files from uaclient.timer.update_messaging import update_motd_messages event = event_logger.get_event_logger() @cli_util.verify_json_format_args @cli_util.assert_root @cli_util.assert_attached() @cli_util.assert_lock_file("pro detach") def action_detach(args, *, cfg, **kwargs) -> int: """Perform the detach action for this machine. @return: 0 on success, 1 otherwise """ ret = _detach( cfg, assume_yes=args.assume_yes, json_output=(args.format == "json") ) if ret == 0: daemon.start() timer.stop() event.process_events() return ret def _detach(cfg: config.UAConfig, assume_yes: bool, json_output: bool) -> int: """Detach the machine from the active Ubuntu Pro subscription, :param cfg: a ``config.UAConfig`` instance :param assume_yes: Assume a yes answer to any prompts requested. In this case, it means automatically disable any service during detach. :param json_output: output should be json only @return: 0 on success, 1 otherwise """ to_disable = [] for ent_name in entitlements.entitlements_disable_order(cfg): try: ent = entitlements.entitlement_factory( cfg=cfg, name=ent_name, ) except exceptions.EntitlementNotFoundError: continue # For detach, we should not consider that a service # cannot be disabled because of dependent services, # since we are going to disable all of them anyway ret, _ = ent.can_disable(ignore_dependent_services=True) if ret: to_disable.append(ent) if to_disable: event.info(messages.DETACH_WILL_DISABLE.pluralize(len(to_disable))) for ent in to_disable: event.info(" {}".format(ent.name)) if not util.prompt_for_confirmation(assume_yes=assume_yes): return 1 for ent in to_disable: perform_disable( ent, cfg, json_output=json_output, assume_yes=assume_yes, update_status=False, ) machine_token_file = machine_token.get_machine_token_file(cfg) machine_token_file.delete() state_files.delete_state_files() update_motd_messages(cfg) event.info(messages.DETACH_SUCCESS) return 0 detach_command = ProCommand( "detach", help=messages.CLI_ROOT_DETACH, description=messages.CLI_DETACH_DESC, action=action_detach, help_category=HelpCategory.OTHER, argument_groups=[ ProArgumentGroup( arguments=[ ProArgument( "--assume-yes", help=messages.CLI_ASSUME_YES.format(command="detach"), action="store_true", ), ProArgument( "--format", help=messages.CLI_FORMAT_DESC.format(default="cli"), action="store", choices=["cli", "json"], default="cli", ), ] ) ], ) help.py 0000644 00000003253 15030160121 0006036 0 ustar 00 import json from uaclient import config, entitlements, messages, status from uaclient.cli.commands import ProArgument, ProArgumentGroup, ProCommand def action_help(args, *, cfg, **kwargs): service = args.service if not service: # Avoiding a circular import from uaclient.cli import get_parser get_parser().print_help() return 0 if not cfg: cfg = config.UAConfig() help_response = status.help(cfg, service) if args.format == "json": print(json.dumps(help_response)) else: for key, value in help_response.items(): print("{}:\n{}\n".format(key.title(), value)) return 0 help_command = ProCommand( "help", help=messages.CLI_ROOT_HELP, description=messages.CLI_HELP_DESC, action=action_help, argument_groups=[ ProArgumentGroup( arguments=[ ProArgument( "service", help=messages.CLI_HELP_SERVICE.format( options=", ".join( entitlements.valid_services(cfg=config.UAConfig()) ) ), action="store", nargs="?", ), ProArgument( "--format", help=(messages.CLI_FORMAT_DESC.format(default="tabular")), action="store", choices=["tabular", "json", "yaml"], default="tabular", ), ProArgument( "--all", help=messages.CLI_HELP_ALL, action="store_true" ), ] ) ], ) attach.py 0000644 00000012753 15030160121 0006357 0 ustar 00 import argparse import sys from uaclient import ( actions, contract, event_logger, exceptions, messages, secret_manager, ) from uaclient.api.u.pro.attach.magic.initiate.v1 import _initiate from uaclient.api.u.pro.attach.magic.revoke.v1 import ( MagicAttachRevokeOptions, _revoke, ) from uaclient.api.u.pro.attach.magic.wait.v1 import ( MagicAttachWaitOptions, _wait, ) from uaclient.cli import cli_util from uaclient.cli.commands import ProArgument, ProArgumentGroup, ProCommand from uaclient.cli.parser import HelpCategory from uaclient.data_types import AttachActionsConfigFile, IncorrectTypeError from uaclient.entitlements import ( create_enable_entitlements_not_found_error, get_valid_entitlement_names, ) from uaclient.entitlements.entitlement_status import CanEnableFailure from uaclient.yaml import safe_load event = event_logger.get_event_logger() def _magic_attach(args, *, cfg, **kwargs): if args.format == "json": raise exceptions.MagicAttachInvalidParam( param="--format", value=args.format, ) event.info(messages.CLI_MAGIC_ATTACH_INIT) initiate_resp = _initiate(cfg=cfg) event.info( "\n" + messages.CLI_MAGIC_ATTACH_SIGN_IN.format( user_code=initiate_resp.user_code ) ) wait_options = MagicAttachWaitOptions(magic_token=initiate_resp.token) try: wait_resp = _wait(options=wait_options, cfg=cfg) except exceptions.MagicAttachTokenError as e: event.info(messages.CLI_MAGIC_ATTACH_FAILED) revoke_options = MagicAttachRevokeOptions( magic_token=initiate_resp.token ) _revoke(options=revoke_options, cfg=cfg) raise e event.info("\n" + messages.CLI_MAGIC_ATTACH_PROCESSING) return wait_resp.contract_token @cli_util.assert_not_attached @cli_util.assert_root @cli_util.assert_lock_file("pro attach") def action_attach(args, *, cfg, **kwargs): if args.token and args.attach_config: raise exceptions.CLIAttachTokenArgXORConfig() elif not args.token and not args.attach_config: token = _magic_attach(args, cfg=cfg) enable_services_override = None elif args.token: token = args.token secret_manager.secrets.add_secret(token) enable_services_override = None else: try: attach_config = AttachActionsConfigFile.from_dict( safe_load(args.attach_config) ) except IncorrectTypeError as e: raise exceptions.AttachInvalidConfigFileError( config_name=args.attach_config.name, error=e.msg ) token = attach_config.token enable_services_override = attach_config.enable_services allow_enable = args.auto_enable and enable_services_override is None try: actions.attach_with_token(cfg, token=token, allow_enable=allow_enable) except exceptions.ConnectivityError: raise exceptions.AttachError() else: ret = 0 if enable_services_override is not None and args.auto_enable: found, not_found = get_valid_entitlement_names( enable_services_override, cfg ) for name in found: ent_ret, reason = actions.enable_entitlement_by_name(cfg, name) if not ent_ret: ret = 1 if ( reason is not None and isinstance(reason, CanEnableFailure) and reason.message is not None ): event.info(reason.message.msg) event.error( error_msg=reason.message.msg, error_code=reason.message.name, service=name, ) else: event.service_processed(name) if not_found: error = create_enable_entitlements_not_found_error( not_found, cfg=cfg ) event.info(error.msg, file_type=sys.stderr) event.error(error_msg=error.msg, error_code=error.msg_code) ret = 1 contract_client = contract.UAContractClient(cfg) contract_client.update_activity_token() cli_util.post_cli_attach(cfg) return ret attach_command = ProCommand( "attach", help=messages.CLI_ROOT_ATTACH, description=messages.CLI_ATTACH_DESC, action=action_attach, preserve_description=True, help_category=HelpCategory.QUICKSTART, help_position=2, argument_groups=[ ProArgumentGroup( arguments=[ ProArgument( "token", help=messages.CLI_ATTACH_TOKEN, nargs="?" ), ProArgument( "--no-auto-enable", help=messages.CLI_ATTACH_NO_AUTO_ENABLE, action="store_false", dest="auto_enable", ), ProArgument( "--attach-config", help=messages.CLI_ATTACH_ATTACH_CONFIG, type=argparse.FileType("r"), ), ProArgument( "--format", help=messages.CLI_FORMAT_DESC.format(default="cli"), action="store", choices=["cli", "json"], default="cli", ), ] ) ], ) fix.py 0000644 00000071320 15030160121 0005674 0 ustar 00 import re import textwrap from typing import ( # noqa: F401 Dict, List, NamedTuple, Optional, Set, Tuple, Union, ) from uaclient import apt, exceptions, messages, system, util from uaclient.actions import attach_with_token, enable_entitlement_by_name from uaclient.api.u.pro.attach.magic.initiate.v1 import _initiate from uaclient.api.u.pro.attach.magic.revoke.v1 import ( MagicAttachRevokeOptions, _revoke, ) from uaclient.api.u.pro.attach.magic.wait.v1 import ( MagicAttachWaitOptions, _wait, ) from uaclient.api.u.pro.security.fix._common import ( CVE_OR_USN_REGEX, FixStatus, UnfixedPackage, status_message, ) from uaclient.api.u.pro.security.fix._common.plan.v1 import ( # noqa: F401 ESM_APPS_POCKET, ESM_INFRA_POCKET, STANDARD_UPDATES_POCKET, FixPlanAptUpgradeStep, FixPlanAttachStep, FixPlanEnableStep, FixPlanNoOpAlreadyFixedStep, FixPlanNoOpLivepatchFixStep, FixPlanNoOpStatus, FixPlanNoOpStep, FixPlanResult, FixPlanStep, FixPlanUSNResult, FixPlanWarning, FixPlanWarningFailUpdatingESMCache, FixPlanWarningPackageCannotBeInstalled, FixPlanWarningSecurityIssueNotFixed, NoOpAlreadyFixedData, NoOpLivepatchFixData, USNAdditionalData, ) from uaclient.api.u.pro.security.fix.cve.plan.v1 import CVEFixPlanOptions from uaclient.api.u.pro.security.fix.cve.plan.v1 import _plan as cve_plan from uaclient.api.u.pro.security.fix.usn.plan.v1 import USNFixPlanOptions from uaclient.api.u.pro.security.fix.usn.plan.v1 import _plan as usn_plan from uaclient.api.u.pro.status.is_attached.v1 import ( ContractExpiryStatus, _is_attached, ) from uaclient.cli.commands import ProArgument, ProArgumentGroup, ProCommand from uaclient.cli.detach import action_detach from uaclient.cli.parser import HelpCategory from uaclient.clouds.identity import ( CLOUD_TYPE_TO_TITLE, PRO_CLOUD_URLS, get_cloud_type, ) from uaclient.config import UAConfig from uaclient.defaults import PRINT_WRAP_WIDTH from uaclient.entitlements import entitlement_factory from uaclient.entitlements.entitlement_status import ( ApplicabilityStatus, CanEnableFailure, UserFacingStatus, ) from uaclient.files import notices from uaclient.files.notices import Notice from uaclient.messages.urls import PRO_HOME_PAGE from uaclient.status import colorize_commands class FixContext: def __init__( self, title: str, dry_run: bool, affected_pkgs: List[str], cfg: UAConfig, ): self.pkg_index = 0 self.unfixed_pkgs = [] # type: List[UnfixedPackage] self.installed_pkgs = set() # type: Set[str] self.fix_status = FixStatus.SYSTEM_NON_VULNERABLE self.title = title self.affected_pkgs = affected_pkgs self.dry_run = dry_run self.cfg = cfg self.should_print_pkg_header = True self.warn_package_cannot_be_installed = False self.fixed_by_livepatch = False def print_fix_header(self): if self.affected_pkgs: msg = messages.SECURITY_AFFECTED_PKGS.pluralize( len(self.affected_pkgs) ).format( count=len(self.affected_pkgs), pkgs=", ".join(sorted(self.affected_pkgs)), ) print( textwrap.fill( msg, width=PRINT_WRAP_WIDTH, subsequent_indent=" ", replace_whitespace=False, ) ) def print_pkg_header( self, source_pkgs: List[str], status: str, pocket: Optional[str] = None, ): if self.should_print_pkg_header: print( _format_packages_message( pkg_list=source_pkgs, status=status, pkg_index=self.pkg_index, num_pkgs=len(self.affected_pkgs), pocket_source=( get_pocket_description(pocket) if pocket else None ), ) ) def add_unfixed_packages(self, pkgs: List[str], unfixed_reason: str): for pkg in pkgs: self.unfixed_pkgs.append( UnfixedPackage(pkg=pkg, unfixed_reason=unfixed_reason) ) def print_cve_header(cve: FixPlanResult): lines = [ "{issue}: {description}".format( issue=cve.title.upper(), description=cve.description ), " - https://ubuntu.com/security/{}".format(cve.title.upper()), ] print("\n".join(lines)) def print_usn_header(fix_plan: FixPlanUSNResult): target_usn = fix_plan.target_usn_plan lines = [ "{issue}: {description}".format( issue=target_usn.title.upper(), description=target_usn.description ), ] additional_data = target_usn.additional_data if isinstance(additional_data, USNAdditionalData): if additional_data.associated_cves: lines.append(messages.SECURITY_FOUND_CVES) for cve in additional_data.associated_cves: lines.append( " - {}".format( messages.urls.SECURITY_CVE_PAGE.format(cve=cve) ) ) elif additional_data.associated_launchpad_bugs: lines.append(messages.SECURITY_FOUND_LAUNCHPAD_BUGS) for lp_bug in additional_data.associated_launchpad_bugs: lines.append(" - " + lp_bug) print("\n".join(lines)) def fix_cve(security_issue: str, dry_run: bool, cfg: UAConfig): fix_plan = cve_plan( options=CVEFixPlanOptions(cves=[security_issue]), cfg=cfg ) error = fix_plan.cves_data.cves[0].error if error and error.msg: raise exceptions.AnonymousUbuntuProError( named_msg=messages.NamedMessage( error.code or "unexpected-error", error.msg ) ) print_cve_header(fix_plan.cves_data.cves[0]) print() status, _ = execute_fix_plan(fix_plan.cves_data.cves[0], dry_run, cfg) return status def fix_usn( security_issue: str, dry_run: bool, no_related: bool, cfg: UAConfig ): fix_plan = usn_plan( options=USNFixPlanOptions(usns=[security_issue]), cfg=cfg ) error = fix_plan.usns_data.usns[0].target_usn_plan.error if error and error.msg: raise exceptions.AnonymousUbuntuProError( named_msg=messages.NamedMessage( error.code or "unexpected-error", error.msg ) ) print_usn_header(fix_plan.usns_data.usns[0]) print( "\n" + messages.SECURITY_FIXING_REQUESTED_USN.format( issue_id=security_issue ) ) target_usn_status, _ = execute_fix_plan( fix_plan.usns_data.usns[0].target_usn_plan, dry_run, cfg, ) if target_usn_status not in ( FixStatus.SYSTEM_NON_VULNERABLE, FixStatus.SYSTEM_NOT_AFFECTED, ): return target_usn_status related_usns_plan = fix_plan.usns_data.usns[0].related_usns_plan if not related_usns_plan or no_related: return target_usn_status print( "\n" + messages.SECURITY_RELATED_USNS.format( related_usns="\n- ".join(usn.title for usn in related_usns_plan) ) ) print("\n" + messages.SECURITY_FIXING_RELATED_USNS) related_usn_status = ( {} ) # type: Dict[str, Tuple[FixStatus, List[UnfixedPackage]]] for related_usn_plan in related_usns_plan: print("- {}".format(related_usn_plan.title)) related_usn_status[related_usn_plan.title] = execute_fix_plan( related_usn_plan, dry_run, cfg, ) print() print(messages.SECURITY_USN_SUMMARY) _handle_fix_status_message( target_usn_status, security_issue, context=messages.FIX_ISSUE_CONTEXT_REQUESTED, ) failure_on_related_usn = False for related_usn_plan in related_usns_plan: status, unfixed_pkgs = related_usn_status[related_usn_plan.title] _handle_fix_status_message( status, related_usn_plan.title, context=messages.FIX_ISSUE_CONTEXT_RELATED, ) if status == FixStatus.SYSTEM_VULNERABLE_UNTIL_REBOOT: print( "- " + messages.ENABLE_REBOOT_REQUIRED_TMPL.format( operation="fix operation" ) ) failure_on_related_usn = True if status == FixStatus.SYSTEM_STILL_VULNERABLE: for unfixed_pkg in unfixed_pkgs: if unfixed_pkg.unfixed_reason: print( " - {}: {}".format( unfixed_pkg.pkg, unfixed_pkg.unfixed_reason ) ) failure_on_related_usn = True if failure_on_related_usn: print( "\n" + messages.SECURITY_RELATED_USN_ERROR.format( issue_id=security_issue ) ) return target_usn_status def _format_packages_message( pkg_list: List[str], status: str, pkg_index: int, num_pkgs: int, pocket_source: Optional[str] = None, ) -> str: """Format the packages and status to an user friendly message.""" if not pkg_list: return "" msg_index = [] src_pkgs = [] for src_pkg in pkg_list: pkg_index += 1 msg_index.append("{}/{}".format(pkg_index, num_pkgs)) src_pkgs.append(src_pkg) msg_header = textwrap.fill( "{} {}:".format( "(" + ", ".join(msg_index) + ")", ", ".join(sorted(src_pkgs)) ), width=PRINT_WRAP_WIDTH, subsequent_indent=" ", ) return "{}\n{}".format(msg_header, status_message(status, pocket_source)) def _run_ua_attach(cfg: UAConfig, token: str) -> bool: """Attach to an Ubuntu Pro subscription with a given token. :return: True if attach performed without errors. """ print(colorize_commands([["pro", "attach", token]])) try: attach_with_token(cfg, token=token, allow_enable=True) return True except exceptions.UbuntuProError as err: print(err.msg) return False def _inform_ubuntu_pro_existence_if_applicable() -> None: """Alert the user when running Pro on cloud with PRO support.""" cloud_type, _ = get_cloud_type() if cloud_type in PRO_CLOUD_URLS.keys(): print( messages.SECURITY_USE_PRO_TMPL.format( title=CLOUD_TYPE_TO_TITLE.get(cloud_type), cloud_specific_url=PRO_CLOUD_URLS.get(cloud_type), ) ) def _perform_magic_attach(cfg: UAConfig): print(messages.CLI_MAGIC_ATTACH_INIT) initiate_resp = _initiate(cfg=cfg) print( "\n" + messages.CLI_MAGIC_ATTACH_SIGN_IN.format( user_code=initiate_resp.user_code ) ) wait_options = MagicAttachWaitOptions(magic_token=initiate_resp.token) try: wait_resp = _wait(options=wait_options, cfg=cfg) except exceptions.MagicAttachTokenError as e: print(messages.CLI_MAGIC_ATTACH_FAILED) revoke_options = MagicAttachRevokeOptions( magic_token=initiate_resp.token ) _revoke(options=revoke_options, cfg=cfg) raise e print("\n" + messages.CLI_MAGIC_ATTACH_PROCESSING) return _run_ua_attach(cfg, wait_resp.contract_token) def _prompt_for_attach(cfg: UAConfig) -> bool: """Prompt for attach to a subscription or token. :return: True if attach performed. """ _inform_ubuntu_pro_existence_if_applicable() print(messages.SECURITY_UPDATE_NOT_INSTALLED_SUBSCRIPTION) choice = util.prompt_choices( messages.SECURITY_FIX_ATTACH_PROMPT, valid_choices=["s", "a", "c"], ) if choice == "c": return False if choice == "s": return _perform_magic_attach(cfg) if choice == "a": print(messages.PROMPT_ENTER_TOKEN) token = input("> ") return _run_ua_attach(cfg, token) return True def _format_unfixed_packages_msg(unfixed_pkgs: List[str]) -> str: """Format the list of unfixed packages into an message. :returns: A string containing the message output for the unfixed packages. """ num_pkgs_unfixed = len(unfixed_pkgs) return textwrap.fill( messages.SECURITY_PKG_STILL_AFFECTED.pluralize( num_pkgs_unfixed ).format( num_pkgs=num_pkgs_unfixed, pkgs=", ".join(sorted(unfixed_pkgs)), ), width=PRINT_WRAP_WIDTH, subsequent_indent=" ", ) def _check_subscription_is_expired(cfg: UAConfig, dry_run: bool) -> bool: """Check if the Ubuntu Pro subscription is expired. :returns: True if subscription is expired and not renewed. """ contract_expiry_status = _is_attached(cfg).contract_status if ( contract_expiry_status and contract_expiry_status == ContractExpiryStatus.EXPIRED.value ): if dry_run: print(messages.SECURITY_DRY_RUN_UA_EXPIRED_SUBSCRIPTION) return False return True return False def _prompt_for_new_token(cfg: UAConfig) -> bool: """Prompt for attach a new subscription token to the user. :return: True if attach performed. """ import argparse _inform_ubuntu_pro_existence_if_applicable() print(messages.SECURITY_UPDATE_NOT_INSTALLED_EXPIRED) choice = util.prompt_choices( messages.SECURITY_FIX_RENEW_PROMPT.format(url=PRO_HOME_PAGE), valid_choices=["r", "c"], ) if choice == "r": print(messages.PROMPT_EXPIRED_ENTER_TOKEN) token = input("> ") print(colorize_commands([["pro", "detach"]])) action_detach(argparse.Namespace(assume_yes=True, format="cli"), cfg) return _run_ua_attach(cfg, token) return False def _prompt_for_enable(cfg: UAConfig, service: str) -> bool: """Prompt for enable a pro service. :return: True if enable performed. """ print(messages.SECURITY_SERVICE_DISABLED.format(service=service)) choice = util.prompt_choices( messages.SECURITY_FIX_ENABLE_PROMPT.format(service=service), valid_choices=["e", "c"], ) if choice == "e": print(colorize_commands([["pro", "enable", service]])) ret, reason = enable_entitlement_by_name(cfg=cfg, name=service) if ( not ret and reason is not None and isinstance(reason, CanEnableFailure) ): if reason.message is not None: print(reason.message.msg) return ret return False def _handle_subscription_for_required_service( service: str, cfg: UAConfig, dry_run: bool ) -> bool: """ Verify if the Ubuntu Pro subscription has the required service enabled. """ ent = entitlement_factory(cfg=cfg, name=service) if ent: ent_status, _ = ent.user_facing_status() if ent_status == UserFacingStatus.ACTIVE: return True applicability_status, _ = ent.applicability_status() if applicability_status == ApplicabilityStatus.APPLICABLE: if dry_run: print( "\n" + messages.SECURITY_DRY_RUN_UA_SERVICE_NOT_ENABLED.format( service=ent.name ) ) return True if _prompt_for_enable(cfg, ent.name): return True else: print( messages.SECURITY_UA_SERVICE_NOT_ENABLED.format( service=ent.name ) ) else: print( messages.SECURITY_UA_SERVICE_NOT_ENTITLED.format( service=ent.name ) ) return False def _handle_fix_status_message( status: FixStatus, issue_id: str, context: str = "" ): if status == FixStatus.SYSTEM_NON_VULNERABLE: if context: msg = messages.SECURITY_ISSUE_RESOLVED_ISSUE_CONTEXT.format( issue=issue_id, context=context ) else: msg = messages.SECURITY_ISSUE_RESOLVED.format(issue=issue_id) print(util.handle_unicode_characters(msg)) elif status == FixStatus.SYSTEM_NOT_AFFECTED: if context: msg = messages.SECURITY_ISSUE_UNAFFECTED_ISSUE_CONTEXT.format( issue=issue_id, context=context ) else: msg = messages.SECURITY_ISSUE_UNAFFECTED.format(issue=issue_id) print(util.handle_unicode_characters(msg)) elif status == FixStatus.SYSTEM_VULNERABLE_UNTIL_REBOOT: if context: msg = messages.SECURITY_ISSUE_NOT_RESOLVED_ISSUE_CONTEXT.format( issue=issue_id, context=context ) else: msg = messages.SECURITY_ISSUE_NOT_RESOLVED.format(issue=issue_id) print(util.handle_unicode_characters(msg)) else: if context: msg = messages.SECURITY_ISSUE_NOT_RESOLVED_ISSUE_CONTEXT.format( issue=issue_id, context=context ) else: msg = messages.SECURITY_ISSUE_NOT_RESOLVED.format(issue=issue_id) print(util.handle_unicode_characters(msg)) def get_pocket_description(pocket: str): if pocket == STANDARD_UPDATES_POCKET: return messages.SECURITY_UBUNTU_STANDARD_UPDATES_POCKET elif pocket == ESM_INFRA_POCKET: return messages.SECURITY_UA_INFRA_POCKET elif pocket == ESM_APPS_POCKET: return messages.SECURITY_UA_APPS_POCKET else: return pocket def _execute_package_cannot_be_installed_step( fix_context: FixContext, step: FixPlanWarningPackageCannotBeInstalled, ): fix_context.print_pkg_header( source_pkgs=step.data.related_source_packages, status="released", pocket=step.data.pocket, ) fix_context.should_print_pkg_header = False warn_msg = messages.FIX_CANNOT_INSTALL_PACKAGE.format( package=step.data.binary_package, version=step.data.binary_package_version, ) print("- " + warn_msg) fix_context.add_unfixed_packages( pkgs=[step.data.source_package], unfixed_reason=warn_msg ) fix_context.warn_package_cannot_be_installed = True fix_context.fix_status = FixStatus.SYSTEM_STILL_VULNERABLE def _execute_security_issue_not_fixed_step( fix_context: FixContext, step: FixPlanWarningSecurityIssueNotFixed ): fix_context.print_pkg_header( source_pkgs=step.data.source_packages, status=step.data.status, ) fix_context.pkg_index += len(step.data.source_packages) fix_context.add_unfixed_packages( pkgs=step.data.source_packages, unfixed_reason=status_message(step.data.status), ) fix_context.fix_status = FixStatus.SYSTEM_STILL_VULNERABLE def _execute_fail_updating_esm_cache_step( fix_context: FixContext, step: FixPlanWarningFailUpdatingESMCache ): if util.we_are_currently_root(): print(messages.CLI_FIX_FAIL_UPDATING_ESM_CACHE) else: print("\n" + messages.CLI_FIX_FAIL_UPDATING_ESM_CACHE_NON_ROOT + "\n") def _execute_apt_upgrade_step( fix_context: FixContext, step: FixPlanAptUpgradeStep, ): fix_context.print_pkg_header( source_pkgs=step.data.source_packages, status="released", pocket=step.data.pocket, ) fix_context.pkg_index += len(step.data.source_packages) if not step.data.binary_packages: if not fix_context.warn_package_cannot_be_installed: print(messages.SECURITY_UPDATE_INSTALLED) fix_context.fix_status = FixStatus.SYSTEM_NON_VULNERABLE return if not util.we_are_currently_root() and not fix_context.dry_run: print(messages.SECURITY_APT_NON_ROOT) fix_context.fix_status = FixStatus.SYSTEM_STILL_VULNERABLE fix_context.add_unfixed_packages( pkgs=step.data.source_packages, unfixed_reason=messages.SECURITY_APT_NON_ROOT, ) return print( colorize_commands( [ ["apt", "update", "&&"] + ["apt", "install", "--only-upgrade", "-y"] + sorted(step.data.binary_packages) ] ) ) if fix_context.dry_run: fix_context.fix_status = FixStatus.SYSTEM_NON_VULNERABLE return try: apt.run_apt_update_command() apt.run_apt_command( cmd=["apt-get", "install", "--only-upgrade", "-y"] + step.data.binary_packages, override_env_vars={"DEBIAN_FRONTEND": "noninteractive"}, ) except Exception as e: msg = getattr(e, "msg", str(e)) print(msg) fix_context.fix_status = FixStatus.SYSTEM_STILL_VULNERABLE fix_context.add_unfixed_packages( pkgs=step.data.source_packages, unfixed_reason=msg, ) return fix_context.fix_status = FixStatus.SYSTEM_NON_VULNERABLE fix_context.should_print_pkg_header = True fix_context.installed_pkgs.update(step.data.binary_packages) def _execute_attach_step( fix_context: FixContext, step: FixPlanAttachStep, ): pocket = ( ESM_INFRA_POCKET if step.data.required_service == "esm-infra" else ESM_APPS_POCKET ) fix_context.print_pkg_header( source_pkgs=step.data.source_packages, status="released", pocket=pocket, ) fix_context.should_print_pkg_header = False if not _is_attached(fix_context.cfg).is_attached: if fix_context.dry_run: print("\n" + messages.SECURITY_DRY_RUN_UA_NOT_ATTACHED) else: if not _prompt_for_attach(fix_context.cfg): fix_context.fix_status = FixStatus.SYSTEM_STILL_VULNERABLE fix_context.add_unfixed_packages( pkgs=step.data.source_packages, unfixed_reason=messages.SECURITY_UA_SERVICE_REQUIRED.format( # noqa service=step.data.required_service ), ) return elif _check_subscription_is_expired( cfg=fix_context.cfg, dry_run=fix_context.dry_run ): if fix_context.dry_run: print(messages.SECURITY_DRY_RUN_UA_EXPIRED_SUBSCRIPTION) elif not _prompt_for_new_token(fix_context.cfg): fix_context.fix_status = FixStatus.SYSTEM_STILL_VULNERABLE fix_context.add_unfixed_packages( pkgs=step.data.source_packages, unfixed_reason=messages.SECURITY_UA_SERVICE_WITH_EXPIRED_SUB.format( # noqa service=step.data.required_service ), ) return fix_context.fix_status = FixStatus.SYSTEM_NON_VULNERABLE def _execute_enable_step( fix_context: FixContext, step: FixPlanEnableStep, ): pocket = ( ESM_INFRA_POCKET if step.data.service == "esm-infra" else ESM_APPS_POCKET ) fix_context.print_pkg_header( source_pkgs=step.data.source_packages, status="released", pocket=pocket, ) fix_context.should_print_pkg_header = False if not _handle_subscription_for_required_service( # noqa step.data.service, fix_context.cfg, fix_context.dry_run, ): fix_context.add_unfixed_packages( pkgs=step.data.source_packages, unfixed_reason=messages.SECURITY_UA_SERVICE_NOT_ENABLED_SHORT.format( # noqa service=step.data.service ), ) fix_context.fix_status = FixStatus.SYSTEM_STILL_VULNERABLE return return FixStatus.SYSTEM_NON_VULNERABLE def _execute_noop_not_affected_step( fix_context: FixContext, step: FixPlanNoOpStep ): if step.data.status == FixPlanNoOpStatus.NOT_AFFECTED.value: print(messages.SECURITY_NO_AFFECTED_PKGS) fix_context.fix_status = FixStatus.SYSTEM_NOT_AFFECTED def _execute_noop_fixed_by_livepatch_step( fix_context: FixContext, step: FixPlanNoOpLivepatchFixStep ): if isinstance(step.data, NoOpLivepatchFixData): print( messages.CVE_FIXED_BY_LIVEPATCH.format( issue=fix_context.title, version=step.data.patch_version, ) ) fix_context.fixed_by_livepatch = True def _execute_noop_already_fixed_step( fix_context: FixContext, step: FixPlanNoOpAlreadyFixedStep ): if isinstance(step.data, NoOpAlreadyFixedData): fix_context.print_pkg_header( source_pkgs=step.data.source_packages, status="released", pocket=step.data.pocket, ) print(messages.SECURITY_UPDATE_INSTALLED) fix_context.pkg_index += len(step.data.source_packages) def execute_fix_plan( fix_plan: FixPlanResult, dry_run: bool, cfg: UAConfig ) -> Tuple[FixStatus, List[UnfixedPackage]]: full_plan = [ *fix_plan.plan, *fix_plan.warnings, ] # type: List[Union[FixPlanStep, FixPlanWarning]] fix_context = FixContext( title=fix_plan.title, dry_run=dry_run, affected_pkgs=fix_plan.affected_packages or [], cfg=cfg, ) fix_context.print_fix_header() for step in sorted(full_plan, key=lambda x: x.order): if isinstance(step, FixPlanWarningPackageCannotBeInstalled): _execute_package_cannot_be_installed_step(fix_context, step) if isinstance(step, FixPlanWarningSecurityIssueNotFixed): _execute_security_issue_not_fixed_step(fix_context, step) if isinstance(step, FixPlanWarningFailUpdatingESMCache): _execute_fail_updating_esm_cache_step(fix_context, step) if isinstance(step, FixPlanAptUpgradeStep): _execute_apt_upgrade_step(fix_context, step) if fix_context.fix_status != FixStatus.SYSTEM_NON_VULNERABLE: break if isinstance(step, FixPlanAttachStep): _execute_attach_step(fix_context, step) if fix_context.fix_status != FixStatus.SYSTEM_NON_VULNERABLE: break if isinstance(step, FixPlanEnableStep): _execute_enable_step(fix_context, step) if fix_context.fix_status != FixStatus.SYSTEM_NON_VULNERABLE: break if isinstance(step, FixPlanNoOpStep): _execute_noop_not_affected_step(fix_context, step) if isinstance(step, FixPlanNoOpLivepatchFixStep): _execute_noop_fixed_by_livepatch_step(fix_context, step) if isinstance(step, FixPlanNoOpAlreadyFixedStep): _execute_noop_already_fixed_step(fix_context, step) print() if fix_context.unfixed_pkgs: print( _format_unfixed_packages_msg( list( set( [ unfixed_pkg.pkg for unfixed_pkg in fix_context.unfixed_pkgs ] ) ) ) ) fix_context.fix_status = FixStatus.SYSTEM_STILL_VULNERABLE if ( fix_context.fix_status == FixStatus.SYSTEM_NON_VULNERABLE and system.should_reboot(installed_pkgs=fix_context.installed_pkgs) ): fix_context.fix_status = FixStatus.SYSTEM_VULNERABLE_UNTIL_REBOOT reboot_msg = messages.ENABLE_REBOOT_REQUIRED_TMPL.format( operation="fix operation" ) print(reboot_msg) notices.add( Notice.ENABLE_REBOOT_REQUIRED, operation="fix operation", ) if not fix_context.fixed_by_livepatch: _handle_fix_status_message(fix_context.fix_status, fix_plan.title) return (fix_context.fix_status, fix_context.unfixed_pkgs) def action_fix(args, *, cfg, **kwargs): if not re.match(CVE_OR_USN_REGEX, args.security_issue): raise exceptions.InvalidSecurityIssueIdFormat( issue=args.security_issue ) if args.dry_run: print(messages.SECURITY_DRY_RUN_WARNING) if "cve" in args.security_issue.lower(): status = fix_cve(args.security_issue, args.dry_run, cfg) else: status = fix_usn( args.security_issue, args.dry_run, args.no_related, cfg ) return status.exit_code fix_command = ProCommand( "fix", help=messages.CLI_ROOT_FIX, description=messages.CLI_FIX_DESC, action=action_fix, help_category=HelpCategory.SECURITY, argument_groups=[ ProArgumentGroup( arguments=[ ProArgument("security_issue", help=messages.CLI_FIX_ISSUE), ProArgument( "--dry-run", help=messages.CLI_FIX_DRY_RUN, action="store_true", ), ProArgument( "--no-related", help=messages.CLI_FIX_NO_RELATED, action="store_true", ), ] ) ], ) commands.py 0000644 00000007265 15030160121 0006716 0 ustar 00 import argparse from typing import Callable, Iterable, Optional, Union from uaclient import messages from uaclient.cli.parser import HelpCategory class ProArgument: def __init__( self, long_name: str, help: str, short_name: Optional[str] = None, **kwargs ): self.names = ( (long_name,) if short_name is None else (short_name, long_name) ) self.help = help self.additional_args = kwargs def register( self, target: Union[argparse.ArgumentParser, argparse._ArgumentGroup] ): target.add_argument( *self.names, help=self.help, **self.additional_args ) class ProArgumentMutuallyExclusiveGroup: def __init__( self, required: bool = False, arguments: Iterable[ProArgument] = (), ): self.required = required self.arguments = arguments class ProArgumentGroup: def __init__( self, title: Optional[str] = None, description: Optional[str] = None, arguments: Iterable[ProArgument] = (), mutually_exclusive_groups: Iterable[ ProArgumentMutuallyExclusiveGroup ] = (), ): self.title = title self.description = description self.arguments = arguments self.mutually_exclusive_groups = mutually_exclusive_groups def register(self, parser: argparse.ArgumentParser): target = ( parser ) # type: Union[argparse.ArgumentParser, argparse._ArgumentGroup] if self.title: target = parser.add_argument_group(self.title, self.description) for argument in self.arguments: argument.register(target) for group in self.mutually_exclusive_groups: new_group = target.add_mutually_exclusive_group( required=group.required ) for argument in group.arguments: argument.register(new_group) class ProCommand: def __init__( self, name: str, help: str, description: str, action: Callable = lambda *args, **kwargs: None, preserve_description: bool = False, argument_groups: Iterable[ProArgumentGroup] = (), subcommands: Iterable["ProCommand"] = (), help_category: Optional[HelpCategory] = None, help_position: int = 0, ): self.name = name self.help = help self.description = description self.action = action self.preserve_description = preserve_description self.argument_groups = argument_groups self.subcommands = subcommands self.help_category = help_category self.help_position = help_position def register(self, subparsers: argparse._SubParsersAction): self.parser = subparsers.add_parser( self.name, help=self.help, description=self.description, ) if self.preserve_description: self.parser.formatter_class = argparse.RawDescriptionHelpFormatter if self.help_category: self.parser.add_help_entry( category=self.help_category, name=self.name, help_string=self.help, position=self.help_position, ) for argument_group in self.argument_groups: argument_group.register(self.parser) self.parser.set_defaults(action=self.action) if self.subcommands: subparsers = self.parser.add_subparsers( title=messages.CLI_AVAILABLE_COMMANDS, dest="command", ) for command in self.subcommands: command.register(subparsers) cli_util.py 0000644 00000013674 15030160121 0006722 0 ustar 00 from functools import wraps from typing import Optional from uaclient import ( actions, api, daemon, entitlements, event_logger, exceptions, lock, messages, status, util, ) from uaclient.api.u.pro.status.is_attached.v1 import _is_attached from uaclient.apt import AptProxyScope, setup_apt_proxy from uaclient.config import UAConfig from uaclient.files import machine_token event = event_logger.get_event_logger() class CLIEnableDisableProgress(api.AbstractProgress): def __init__(self, *, assume_yes: bool): self.is_interactive = not assume_yes self.assume_yes = assume_yes def progress( self, *, total_steps: int, done_steps: int, previous_step_message: Optional[str], current_step_message: Optional[str] ): if current_step_message is not None: print(current_step_message) def _on_event(self, event: str, payload): if event == "info": print(payload) elif event == "message_operation": if not util.handle_message_operations(payload, self.assume_yes): raise exceptions.PromptDeniedError() def _null_print(*args, **kwargs): pass def create_interactive_only_print_function(json_output: bool): if json_output: return _null_print else: return print def assert_lock_file(lock_holder=None): """Decorator asserting exclusive access to lock file""" def wrapper(f): @wraps(f) def new_f(*args, cfg, **kwargs): with lock.RetryLock(lock_holder=lock_holder, sleep_time=1): retval = f(*args, cfg=cfg, **kwargs) return retval return new_f return wrapper def assert_root(f): """Decorator asserting root user""" @wraps(f) def new_f(*args, **kwargs): if not util.we_are_currently_root(): raise exceptions.NonRootUserError() else: return f(*args, **kwargs) return new_f def verify_json_format_args(f): """Decorator to verify if correct params are used for json format""" @wraps(f) def new_f(cmd_args, *args, **kwargs): if not cmd_args: return f(cmd_args, *args, **kwargs) if cmd_args.format == "json" and not cmd_args.assume_yes: raise exceptions.CLIJSONFormatRequireAssumeYes() else: return f(cmd_args, *args, **kwargs) return new_f def assert_attached(raise_custom_error_function=None): """Decorator asserting attached config. :param msg_function: Optional function to generate a custom message if raising an UnattachedError """ def wrapper(f): @wraps(f) def new_f(args, cfg, **kwargs): if not _is_attached(cfg).is_attached: if raise_custom_error_function: command = getattr(args, "command", "") service_names = getattr(args, "service", "") raise_custom_error_function( command=command, service_names=service_names, cfg=cfg ) else: raise exceptions.UnattachedError() return f(args, cfg=cfg, **kwargs) return new_f return wrapper def assert_not_attached(f): """Decorator asserting unattached config.""" @wraps(f) def new_f(args, cfg, **kwargs): if _is_attached(cfg).is_attached: machine_token_file = machine_token.get_machine_token_file() raise exceptions.AlreadyAttachedError( account_name=machine_token_file.account.get("name", "") ) return f(args, cfg=cfg, **kwargs) return new_f def _raise_enable_disable_unattached_error(command, service_names, cfg): """Raises a custom error for enable/disable commands when unattached. Takes into consideration if the services exist or not, and notify the user accordingly.""" ( entitlements_found, entitlements_not_found, ) = entitlements.get_valid_entitlement_names(names=service_names, cfg=cfg) if entitlements_found and entitlements_not_found: raise exceptions.UnattachedMixedServicesError( valid_service=", ".join(entitlements_found), operation=command, invalid_service=", ".join(entitlements_not_found), service_msg="", ) elif entitlements_found: raise exceptions.UnattachedValidServicesError( valid_service=", ".join(entitlements_found), operation=command ) else: raise exceptions.UnattachedInvalidServicesError( operation=command, invalid_service=", ".join(entitlements_not_found), service_msg="", ) def post_cli_attach(cfg: UAConfig) -> None: machine_token_file = machine_token.get_machine_token_file(cfg) contract_name = machine_token_file.contract_name if contract_name: event.info( messages.ATTACH_SUCCESS_TMPL.format(contract_name=contract_name) ) else: event.info(messages.ATTACH_SUCCESS_NO_CONTRACT_NAME) daemon.stop() daemon.cleanup(cfg) status_dict, _ret = actions.status(cfg) output = status.format_tabular(status_dict) event.info(util.handle_unicode_characters(output)) event.process_events() def configure_apt_proxy( cfg: UAConfig, scope: AptProxyScope, set_key: str, set_value: Optional[str], ) -> None: """ Handles setting part the apt proxies - global and uaclient scoped proxies """ if scope == AptProxyScope.GLOBAL: http_proxy = cfg.global_apt_http_proxy https_proxy = cfg.global_apt_https_proxy elif scope == AptProxyScope.UACLIENT: http_proxy = cfg.ua_apt_http_proxy https_proxy = cfg.ua_apt_https_proxy if "https" in set_key: https_proxy = set_value else: http_proxy = set_value setup_apt_proxy( http_proxy=http_proxy, https_proxy=https_proxy, proxy_scope=scope ) convert.py 0000644 00000030532 15030164337 0006603 0 ustar 00 from __future__ import annotations import os.path import re from abc import ABCMeta, abstractmethod from collections import defaultdict from collections.abc import Iterator from email.message import Message from email.parser import Parser from email.policy import EmailPolicy from glob import iglob from pathlib import Path from textwrap import dedent from zipfile import ZipFile from .. import __version__ from ..metadata import generate_requirements from ..vendored.packaging.tags import parse_tag from ..wheelfile import WheelFile egg_filename_re = re.compile( r""" (?P<name>.+?)-(?P<ver>.+?) (-(?P<pyver>py\d\.\d+) (-(?P<arch>.+?))? )?.egg$""", re.VERBOSE, ) egg_info_re = re.compile( r""" ^(?P<name>.+?)-(?P<ver>.+?) (-(?P<pyver>py\d\.\d+) )?.egg-info/""", re.VERBOSE, ) wininst_re = re.compile( r"\.(?P<platform>win32|win-amd64)(?:-(?P<pyver>py\d\.\d))?\.exe$" ) pyd_re = re.compile(r"\.(?P<abi>[a-z0-9]+)-(?P<platform>win32|win_amd64)\.pyd$") serialization_policy = EmailPolicy( utf8=True, mangle_from_=False, max_line_length=0, ) GENERATOR = f"wheel {__version__}" def convert_requires(requires: str, metadata: Message) -> None: extra: str | None = None requirements: dict[str | None, list[str]] = defaultdict(list) for line in requires.splitlines(): line = line.strip() if not line: continue if line.startswith("[") and line.endswith("]"): extra = line[1:-1] continue requirements[extra].append(line) for key, value in generate_requirements(requirements): metadata.add_header(key, value) def convert_pkg_info(pkginfo: str, metadata: Message): parsed_message = Parser().parsestr(pkginfo) for key, value in parsed_message.items(): key_lower = key.lower() if value == "UNKNOWN": continue if key_lower == "description": description_lines = value.splitlines() value = "\n".join( ( description_lines[0].lstrip(), dedent("\n".join(description_lines[1:])), "\n", ) ) metadata.set_payload(value) elif key_lower == "home-page": metadata.add_header("Project-URL", f"Homepage, {value}") elif key_lower == "download-url": metadata.add_header("Project-URL", f"Download, {value}") else: metadata.add_header(key, value) metadata.replace_header("Metadata-Version", "2.4") def normalize(name: str) -> str: return re.sub(r"[-_.]+", "-", name).lower().replace("-", "_") class ConvertSource(metaclass=ABCMeta): name: str version: str pyver: str = "py2.py3" abi: str = "none" platform: str = "any" metadata: Message @property def dist_info_dir(self) -> str: return f"{self.name}-{self.version}.dist-info" @abstractmethod def generate_contents(self) -> Iterator[tuple[str, bytes]]: pass class EggFileSource(ConvertSource): def __init__(self, path: Path): if not (match := egg_filename_re.match(path.name)): raise ValueError(f"Invalid egg file name: {path.name}") # Binary wheels are assumed to be for CPython self.path = path self.name = normalize(match.group("name")) self.version = match.group("ver") if pyver := match.group("pyver"): self.pyver = pyver.replace(".", "") if arch := match.group("arch"): self.abi = self.pyver.replace("py", "cp") self.platform = normalize(arch) self.metadata = Message() def generate_contents(self) -> Iterator[tuple[str, bytes]]: with ZipFile(self.path, "r") as zip_file: for filename in sorted(zip_file.namelist()): # Skip pure directory entries if filename.endswith("/"): continue # Handle files in the egg-info directory specially, selectively moving # them to the dist-info directory while converting as needed if filename.startswith("EGG-INFO/"): if filename == "EGG-INFO/requires.txt": requires = zip_file.read(filename).decode("utf-8") convert_requires(requires, self.metadata) elif filename == "EGG-INFO/PKG-INFO": pkginfo = zip_file.read(filename).decode("utf-8") convert_pkg_info(pkginfo, self.metadata) elif filename == "EGG-INFO/entry_points.txt": yield ( f"{self.dist_info_dir}/entry_points.txt", zip_file.read(filename), ) continue # For any other file, just pass it through yield filename, zip_file.read(filename) class EggDirectorySource(EggFileSource): def generate_contents(self) -> Iterator[tuple[str, bytes]]: for dirpath, _, filenames in os.walk(self.path): for filename in sorted(filenames): path = Path(dirpath, filename) if path.parent.name == "EGG-INFO": if path.name == "requires.txt": requires = path.read_text("utf-8") convert_requires(requires, self.metadata) elif path.name == "PKG-INFO": pkginfo = path.read_text("utf-8") convert_pkg_info(pkginfo, self.metadata) if name := self.metadata.get("Name"): self.name = normalize(name) if version := self.metadata.get("Version"): self.version = version elif path.name == "entry_points.txt": yield ( f"{self.dist_info_dir}/entry_points.txt", path.read_bytes(), ) continue # For any other file, just pass it through yield str(path.relative_to(self.path)), path.read_bytes() class WininstFileSource(ConvertSource): """ Handles distributions created with ``bdist_wininst``. The egginfo filename has the format:: name-ver(-pyver)(-arch).egg-info The installer filename has the format:: name-ver.arch(-pyver).exe Some things to note: 1. The installer filename is not definitive. An installer can be renamed and work perfectly well as an installer. So more reliable data should be used whenever possible. 2. The egg-info data should be preferred for the name and version, because these come straight from the distutils metadata, and are mandatory. 3. The pyver from the egg-info data should be ignored, as it is constructed from the version of Python used to build the installer, which is irrelevant - the installer filename is correct here (even to the point that when it's not there, any version is implied). 4. The architecture must be taken from the installer filename, as it is not included in the egg-info data. 5. Architecture-neutral installers still have an architecture because the installer format itself (being executable) is architecture-specific. We should therefore ignore the architecture if the content is pure-python. """ def __init__(self, path: Path): self.path = path self.metadata = Message() # Determine the initial architecture and Python version from the file name # (if possible) if match := wininst_re.search(path.name): self.platform = normalize(match.group("platform")) if pyver := match.group("pyver"): self.pyver = pyver.replace(".", "") # Look for an .egg-info directory and any .pyd files for more precise info egg_info_found = pyd_found = False with ZipFile(self.path) as zip_file: for filename in zip_file.namelist(): prefix, filename = filename.split("/", 1) if not egg_info_found and (match := egg_info_re.match(filename)): egg_info_found = True self.name = normalize(match.group("name")) self.version = match.group("ver") if pyver := match.group("pyver"): self.pyver = pyver.replace(".", "") elif not pyd_found and (match := pyd_re.search(filename)): pyd_found = True self.abi = match.group("abi") self.platform = match.group("platform") if egg_info_found and pyd_found: break def generate_contents(self) -> Iterator[tuple[str, bytes]]: dist_info_dir = f"{self.name}-{self.version}.dist-info" data_dir = f"{self.name}-{self.version}.data" with ZipFile(self.path, "r") as zip_file: for filename in sorted(zip_file.namelist()): # Skip pure directory entries if filename.endswith("/"): continue # Handle files in the egg-info directory specially, selectively moving # them to the dist-info directory while converting as needed prefix, target_filename = filename.split("/", 1) if egg_info_re.search(target_filename): basename = target_filename.rsplit("/", 1)[-1] if basename == "requires.txt": requires = zip_file.read(filename).decode("utf-8") convert_requires(requires, self.metadata) elif basename == "PKG-INFO": pkginfo = zip_file.read(filename).decode("utf-8") convert_pkg_info(pkginfo, self.metadata) elif basename == "entry_points.txt": yield ( f"{dist_info_dir}/entry_points.txt", zip_file.read(filename), ) continue elif prefix == "SCRIPTS": target_filename = f"{data_dir}/scripts/{target_filename}" # For any other file, just pass it through yield target_filename, zip_file.read(filename) def convert(files: list[str], dest_dir: str, verbose: bool) -> None: for pat in files: for archive in iglob(pat): path = Path(archive) if path.suffix == ".egg": if path.is_dir(): source: ConvertSource = EggDirectorySource(path) else: source = EggFileSource(path) else: source = WininstFileSource(path) if verbose: print(f"{archive}...", flush=True, end="") dest_path = Path(dest_dir) / ( f"{source.name}-{source.version}-{source.pyver}-{source.abi}" f"-{source.platform}.whl" ) with WheelFile(dest_path, "w") as wheelfile: for name_or_zinfo, contents in source.generate_contents(): wheelfile.writestr(name_or_zinfo, contents) # Write the METADATA file wheelfile.writestr( f"{source.dist_info_dir}/METADATA", source.metadata.as_string(policy=serialization_policy).encode( "utf-8" ), ) # Write the WHEEL file wheel_message = Message() wheel_message.add_header("Wheel-Version", "1.0") wheel_message.add_header("Generator", GENERATOR) wheel_message.add_header( "Root-Is-Purelib", str(source.platform == "any").lower() ) tags = parse_tag(f"{source.pyver}-{source.abi}-{source.platform}") for tag in sorted(tags, key=lambda tag: tag.interpreter): wheel_message.add_header("Tag", str(tag)) wheelfile.writestr( f"{source.dist_info_dir}/WHEEL", wheel_message.as_string(policy=serialization_policy).encode( "utf-8" ), ) if verbose: print("OK") unpack.py 0000644 00000001775 15030164337 0006413 0 ustar 00 from __future__ import annotations from pathlib import Path from ..wheelfile import WheelFile def unpack(path: str, dest: str = ".") -> None: """Unpack a wheel. Wheel content will be unpacked to {dest}/{name}-{ver}, where {name} is the package name and {ver} its version. :param path: The path to the wheel. :param dest: Destination directory (default to current directory). """ with WheelFile(path) as wf: namever = wf.parsed_filename.group("namever") destination = Path(dest) / namever print(f"Unpacking to: {destination}...", end="", flush=True) for zinfo in wf.filelist: wf.extract(zinfo, destination) # Set permissions to the same values as they were set in the archive # We have to do this manually due to # https://github.com/python/cpython/issues/59999 permissions = zinfo.external_attr >> 16 & 0o777 destination.joinpath(zinfo.filename).chmod(permissions) print("OK") pack.py 0000644 00000006037 15030164337 0006044 0 ustar 00 from __future__ import annotations import email.policy import os.path import re from email.generator import BytesGenerator from email.parser import BytesParser from wheel.cli import WheelError from wheel.wheelfile import WheelFile DIST_INFO_RE = re.compile(r"^(?P<namever>(?P<name>.+?)-(?P<ver>\d.*?))\.dist-info$") def pack(directory: str, dest_dir: str, build_number: str | None) -> None: """Repack a previously unpacked wheel directory into a new wheel file. The .dist-info/WHEEL file must contain one or more tags so that the target wheel file name can be determined. :param directory: The unpacked wheel directory :param dest_dir: Destination directory (defaults to the current directory) """ # Find the .dist-info directory dist_info_dirs = [ fn for fn in os.listdir(directory) if os.path.isdir(os.path.join(directory, fn)) and DIST_INFO_RE.match(fn) ] if len(dist_info_dirs) > 1: raise WheelError(f"Multiple .dist-info directories found in {directory}") elif not dist_info_dirs: raise WheelError(f"No .dist-info directories found in {directory}") # Determine the target wheel filename dist_info_dir = dist_info_dirs[0] name_version = DIST_INFO_RE.match(dist_info_dir).group("namever") # Read the tags and the existing build number from .dist-info/WHEEL wheel_file_path = os.path.join(directory, dist_info_dir, "WHEEL") with open(wheel_file_path, "rb") as f: info = BytesParser(policy=email.policy.compat32).parse(f) tags: list[str] = info.get_all("Tag", []) existing_build_number = info.get("Build") if not tags: raise WheelError( f"No tags present in {dist_info_dir}/WHEEL; cannot determine target " f"wheel filename" ) # Set the wheel file name and add/replace/remove the Build tag in .dist-info/WHEEL build_number = build_number if build_number is not None else existing_build_number if build_number is not None: del info["Build"] if build_number: info["Build"] = build_number name_version += "-" + build_number if build_number != existing_build_number: with open(wheel_file_path, "wb") as f: BytesGenerator(f, maxheaderlen=0).flatten(info) # Reassemble the tags for the wheel file tagline = compute_tagline(tags) # Repack the wheel wheel_path = os.path.join(dest_dir, f"{name_version}-{tagline}.whl") with WheelFile(wheel_path, "w") as wf: print(f"Repacking wheel as {wheel_path}...", end="", flush=True) wf.write_files(directory) print("OK") def compute_tagline(tags: list[str]) -> str: """Compute a tagline from a list of tags. :param tags: A list of tags :return: A tagline """ impls = sorted({tag.split("-")[0] for tag in tags}) abivers = sorted({tag.split("-")[1] for tag in tags}) platforms = sorted({tag.split("-")[2] for tag in tags}) return "-".join([".".join(impls), ".".join(abivers), ".".join(platforms)]) tags.py 0000644 00000011230 15030164337 0006053 0 ustar 00 from __future__ import annotations import email.policy import itertools import os from collections.abc import Iterable from email.parser import BytesParser from ..wheelfile import WheelFile def _compute_tags(original_tags: Iterable[str], new_tags: str | None) -> set[str]: """Add or replace tags. Supports dot-separated tags""" if new_tags is None: return set(original_tags) if new_tags.startswith("+"): return {*original_tags, *new_tags[1:].split(".")} if new_tags.startswith("-"): return set(original_tags) - set(new_tags[1:].split(".")) return set(new_tags.split(".")) def tags( wheel: str, python_tags: str | None = None, abi_tags: str | None = None, platform_tags: str | None = None, build_tag: str | None = None, remove: bool = False, ) -> str: """Change the tags on a wheel file. The tags are left unchanged if they are not specified. To specify "none", use ["none"]. To append to the previous tags, a tag should start with a "+". If a tag starts with "-", it will be removed from existing tags. Processing is done left to right. :param wheel: The paths to the wheels :param python_tags: The Python tags to set :param abi_tags: The ABI tags to set :param platform_tags: The platform tags to set :param build_tag: The build tag to set :param remove: Remove the original wheel """ with WheelFile(wheel, "r") as f: assert f.filename, f"{f.filename} must be available" wheel_info = f.read(f.dist_info_path + "/WHEEL") info = BytesParser(policy=email.policy.compat32).parsebytes(wheel_info) original_wheel_name = os.path.basename(f.filename) namever = f.parsed_filename.group("namever") build = f.parsed_filename.group("build") original_python_tags = f.parsed_filename.group("pyver").split(".") original_abi_tags = f.parsed_filename.group("abi").split(".") original_plat_tags = f.parsed_filename.group("plat").split(".") tags: list[str] = info.get_all("Tag", []) existing_build_tag = info.get("Build") impls = {tag.split("-")[0] for tag in tags} abivers = {tag.split("-")[1] for tag in tags} platforms = {tag.split("-")[2] for tag in tags} if impls != set(original_python_tags): msg = f"Wheel internal tags {impls!r} != filename tags {original_python_tags!r}" raise AssertionError(msg) if abivers != set(original_abi_tags): msg = f"Wheel internal tags {abivers!r} != filename tags {original_abi_tags!r}" raise AssertionError(msg) if platforms != set(original_plat_tags): msg = ( f"Wheel internal tags {platforms!r} != filename tags {original_plat_tags!r}" ) raise AssertionError(msg) if existing_build_tag != build: msg = ( f"Incorrect filename '{build}' " f"& *.dist-info/WHEEL '{existing_build_tag}' build numbers" ) raise AssertionError(msg) # Start changing as needed if build_tag is not None: build = build_tag final_python_tags = sorted(_compute_tags(original_python_tags, python_tags)) final_abi_tags = sorted(_compute_tags(original_abi_tags, abi_tags)) final_plat_tags = sorted(_compute_tags(original_plat_tags, platform_tags)) final_tags = [ namever, ".".join(final_python_tags), ".".join(final_abi_tags), ".".join(final_plat_tags), ] if build: final_tags.insert(1, build) final_wheel_name = "-".join(final_tags) + ".whl" if original_wheel_name != final_wheel_name: del info["Tag"], info["Build"] for a, b, c in itertools.product( final_python_tags, final_abi_tags, final_plat_tags ): info["Tag"] = f"{a}-{b}-{c}" if build: info["Build"] = build original_wheel_path = os.path.join( os.path.dirname(f.filename), original_wheel_name ) final_wheel_path = os.path.join(os.path.dirname(f.filename), final_wheel_name) with WheelFile(original_wheel_path, "r") as fin, WheelFile( final_wheel_path, "w" ) as fout: fout.comment = fin.comment # preserve the comment for item in fin.infolist(): if item.is_dir(): continue if item.filename == f.dist_info_path + "/RECORD": continue if item.filename == f.dist_info_path + "/WHEEL": fout.writestr(item, info.as_bytes()) else: fout.writestr(item, fin.read(item)) if remove: os.remove(original_wheel_path) return final_wheel_name
| ver. 1.4 |
Github
|
.
| PHP 8.2.28 | Generation time: 0.03 |
proxy
|
phpinfo
|
Settings