File manager - Edit - /home/newsbmcs.com/public_html/static/img/logo/_common.tar
Back
__pycache__/__init__.cpython-310.pyc 0000644 00000051551 15030137052 0013217 0 ustar 00 o �P�f�h � @ sZ d dl Z d dlZd dlZd dlmZ d dlmZmZmZm Z m Z m Z d dl m Z mZmZmZmZmZ d dlmZ dZdZdZd Zd Zd Zd Zd Z e dde fde fde fg�Z e dde fde fg�Z G dd� dej �Z!G dd� dej"�Z#G dd� d�Z$G dd � d �Z%G d d � d �Z&d e%d ee ee e f f d!ee e$f fd"d#�Z'd!ee ee e f f fd$d%�Z(d&d'� Z)d(d)� Z*d*d+� Z+d,e&d ee ee e f f d!ee e$f fd-d.�Z,d/e$d0ee ee e f f d!e$fd1d2�Z-d3d4� Z.d5ee& d6ee e/f d!ee ee ee e f f f fd7d8�Z0d9e d!e e e! e e f fd:d;�Z1dCd<e e fd=d>�Z2d?e d@e d!e fdAdB�Z3dS )D� N)� defaultdict)�Any�Dict�List� NamedTuple�Optional�Tuple)�apt� exceptions� livepatch�messages�system�util)� serviceclientz=((CVE|cve)-\d{4}-\d{4,7}$|(USN|usn|LSN|lsn)-\d{1,5}-\d{1,2}$)z cves.jsonzcves/{cve}.jsonz notices.jsonznotices/{notice}.jsonzstandard-updates� esm-infra�esm-apps�BinaryPackageFix� source_pkg� binary_pkg� fixed_version�UnfixedPackage�pkg�unfixed_reasonc @ sZ e Zd ZdZG dd� d�Zedd�Zedd�Zedd�Zed d �Ze d d � �Z d d� Z dS )� FixStatuszD An enum to represent the system status after fix operation c @ s e Zd Zdedefdd�ZdS )zFixStatus._Value�value�msgc C s || _ || _d S �N�r r )�selfr r � r �R/usr/lib/python3/dist-packages/uaclient/api/u/pro/security/fix/_common/__init__.py�__init__/ � zFixStatus._Value.__init__N)�__name__� __module__� __qualname__�int�strr! r r r r �_Value. s r( r �fixed� not-affected� zstill-affected� zaffected-until-rebootc C s | j j S r )r �r r r r � exit_code8 s zFixStatus.exit_codec C s | j jS r r r- r r r �__str__< s zFixStatus.__str__N) r# r$ r% �__doc__r( �SYSTEM_NON_VULNERABLE�SYSTEM_NOT_AFFECTED�SYSTEM_STILL_VULNERABLE� SYSTEM_VULNERABLE_UNTIL_REBOOT�propertyr. r/ r r r r r ) s r c s0 e Zd ZdZdZdeeef deeef fdd�Ze j e j g d�d� d#� fd d � �Z d$d ee d ee dee dee dee dee dee deee ded fdd�Zdeddfdd�Z d%dee dee dee dee d ee ded f d d �Zd edd fd!d"�Z� ZS )&�UASecurityClient� � security_url� query_params�returnc C s. | j j �di ��di �}|r|�|� |S |S )zD Update query params with data from feature config. �features�extra_security_params)�cfg�get�update)r r9 r< r r r �_get_query_paramsE s � z"UASecurityClient._get_query_params)r+ � � )� retry_sleepsNc s" | � |�}t� j|||||dd�S )NF)�path�data�headers�methodr9 �log_response_body)r@ �super� request_url)r rD rE rF rG r9 �� __class__r r rJ U s �z UASecurityClient.request_url�query�priority�package�limit�offset� component�version�status�CVEc sV ||||||||d�} � j t| d�} | jdkr!tjt| j| jd��� fdd�| jD �S )znQuery to match multiple-CVEs. @return: List of CVE instances based on the the JSON response. )�qrN rO rP rQ rR rS rT �r9 �� ��url�code�bodyc s g | ]}t � |d ��qS )��client�response)rU )�.0�cve_mdr- r r � <listcomp>� s �z-UASecurityClient.get_cves.<locals>.<listcomp>)rJ � API_V1_CVESr[ r �SecurityAPIErrorr\ � json_list) r rM rN rO rP rQ rR rS rT r9 r_ r r- r �get_cvesc s" � � �zUASecurityClient.get_cves�cve_idc C �B t j|d�}| �|�}|jdkrtj||j|jd��t| |jd�S )zkQuery to match single-CVE. @return: CVE instance for JSON response from the Security API. )�cverX rY r] ) �API_V1_CVE_TMPL�formatrJ r[ r rd r\ rU � json_dict)r rg rZ r_ r r r �get_cve� � �zUASecurityClient.get_cve�cves�release�order�USNc sf � ||||d�}�j t|d�}|jdkr tjt|j|jd��t� �fdd�|j�dg �D �dd � d �S ) zuQuery to match multiple-USNs. @return: Sorted list of USN instances based on the the JSON response. )ro rp rP rQ rq rW rX rY c sB g | ] }� d u s� |� dg �v r|� dd��d�rt�|d��qS )N�cves_ids�id� �USN-r] )r> � startswithrr )r` �usn_md�ro r r r rb � s � �z0UASecurityClient.get_notices.<locals>.<listcomp>�noticesc S � | j S r �rt ��xr r r �<lambda>� � z.UASecurityClient.get_notices.<locals>.<lambda>��key) rJ �API_V1_NOTICESr[ r rd r\ �sortedrl r> )r ro rp rP rQ rq r9 r_ r ry r � get_notices� s" � � ��z UASecurityClient.get_notices� notice_idc C rh )zbQuery to match single-USN. @return: USN instance representing the JSON response. )�noticerX rY r] ) �API_V1_NOTICE_TMPLrk rJ r[ r rd r\ rr rl )r r� rZ r_ r r r � get_notice� rn zUASecurityClient.get_notice)NNNN)NNNNNNNN)NNNNN)r# r$ r% � url_timeout�cfg_url_base_attrr r' r r@ r �retry�socket�timeoutrJ r r& r rf rm r� r� � __classcell__r r rK r r6 @ st � ���������� � �"������ �#r6 c @ s� e Zd ZdZdeeef fdd�Zedd� �Z edd� �Z ed d � �Z ed d � �Z ed d� �Z edd� �Zedefdd��Zedd� �ZdS )�CVEPackageStatuszAClass representing specific CVE PackageStatus on an Ubuntu series� cve_responsec C s || _ d S r �r_ )r r� r r r r! � s zCVEPackageStatus.__init__c C � | j d S �N� descriptionr� r- r r r r� � � z CVEPackageStatus.descriptionc C r{ r )r� r- r r r r � s z CVEPackageStatus.fixed_versionc C r� )N�pocketr� r- r r r r� � r� zCVEPackageStatus.pocketc C r� )N�release_codenamer� r- r r r r� � r� z!CVEPackageStatus.release_codenamec C r� )NrT r� r- r r r rT � r� zCVEPackageStatus.statusc C s� | j dkrtjS | j dkrtjS | j dkrtjS | j dv r tjS | j dkr(tjS | j dkr0tjS | j dkr=tjj | j d�S tj j | j d �S � N�neededz needs-triage�pending)�ignored�deferred�DNEr* �released)� fix_stream)rT ) rT r �SECURITY_CVE_STATUS_NEEDED�SECURITY_CVE_STATUS_TRIAGE�SECURITY_CVE_STATUS_PENDING�SECURITY_CVE_STATUS_IGNORED�SECURITY_CVE_STATUS_DNE� SECURITY_CVE_STATUS_NOT_AFFECTED�SECURITY_FIX_RELEASE_STREAMrk � pocket_source�SECURITY_CVE_STATUS_UNKNOWNr- r r r �status_message� s" �z CVEPackageStatus.status_messager: c C s t | jtjk�S )z?Return True if the package requires an active Pro subscription.)�boolr� r �'SECURITY_UBUNTU_STANDARD_UPDATES_POCKETr- r r r � requires_ua� s ��z CVEPackageStatus.requires_uac C sZ | j dkr tj}|S | j dkrtj}|S | j dv r tj}|S d| jv r(tj}|S tj}|S )z>Human-readable string representing where the fix is published.r r )�updates�security�esm)r� r �SECURITY_UA_INFRA_POCKET�SECURITY_UA_APPS_POCKETr� r )r � fix_sourcer r r r� � s � � ��z CVEPackageStatus.pocket_sourceN)r# r$ r% r0 r r' r r! r5 r� r r� r� rT r� r� r� r� r r r r r� � s& r� c @ s� e Zd ZdZdedeeef fdd�Zde fdd�Z e d d � �Z e de e fd d ��Ze de d fdd��Ze dd� �Ze deeef fdd��ZdS )rU z7Class representing CVE response from the SecurityClientr^ r_ c C � || _ || _d S r �r_ r^ �r r^ r_ r r r r! r" z CVE.__init__r: c C � t |t�sdS | j|jkS �NF)� isinstancerU r_ �r �otherr r r �__eq__ � z CVE.__eq__c C � | j �dd��� S )Nrt �UNKNOWN_CVE_ID�r_ r> �upperr- r r r rt � zCVE.idc C � | j �dg �S )N� notices_ids�r_ r> r- r r r r� s zCVE.notices_idsrr c �< t � d�st� fdd�� j�dg �D �dd� dd�� _� jS ) z�Return a list of USN instances from API response 'notices'. Cache the value to avoid extra work on multiple calls. �_noticesc s. g | ]}|r|� d d��d�rt� j|��qS )rt ru rv )r> rw rr r^ )r` r� r- r r rb ( s �� �z CVE.notices.<locals>.<listcomp>rz c S r{ r r| ��nr r r r - r� z CVE.notices.<locals>.<lambda>T�r� �reverse)�hasattrr� r_ r> r� r- r r- r rz s �� z CVE.noticesc C � | j �d�S r� r� r- r r r r� 2 � zCVE.descriptionc C sb t | d�r| jS i | _t�� j}| jd D ]}|d D ]}|d |kr,t|�| j|d < qq| jS )z�Dict of package status dicts for the current Ubuntu series. Top-level keys are source packages names and each value is a CVEPackageStatus object �_packages_status�packages�statusesr� �name)r� r� r �get_release_info�seriesr_ r� )r r� rO � pkg_statusr r r �packages_status6 s ���zCVE.packages_statusN)r# r$ r% r0 r6 r r' r r! r� r� r5 rt r r� rz r� r� r� r r r r rU s rU c @ s� e Zd ZdZdedeeef fdd�Zde fdd�Z e defd d ��Z e de e fd d ��Ze de e fd d��Ze dd� �Ze dd� �Ze deeeeeeef f f fdd��ZdS )rr z7Class representing USN response from the SecurityClientr^ r_ c C r� r r� r� r r r r! M r" z USN.__init__r: c C r� r� )r� rr r_ r� r r r r� Q r� z USN.__eq__c C r� )Nrt �UNKNOWN_USN_IDr� r- r r r rt V r� zUSN.idc C r� )z$List of CVE IDs related to this USN.rs r� r- r r r rs Z s z USN.cves_idsc r� ) z�List of CVE instances based on API response 'cves' key. Cache the values to avoid extra work for multiple call-sites. �_cvesc s g | ]}t � j|��qS r )rU r^ )r` ri r- r r rb g s ��z USN.cves.<locals>.<listcomp>ro c S r{ r r| r� r r r r k r� zUSN.cves.<locals>.<lambda>Tr� )r� r� r_ r> r� r- r r- r ro _ s ��zUSN.cvesc C r� )N�titler� r- r r r r� p r� z USN.titlec C r� )N� referencesr� r- r r r r� t r� zUSN.referencesc C sN t | d�r| jS t�� j}i | _| j�di ��|g �D ]�}|�d�rW|d | jv rMd| j|d v rCtjdj | j |d d�| j dd ��|| j|d d< qd|i| j|d < q|�d �smtjd j | j |d d �| j dd ��d |d vr�tjdj | j |d |d d�| j dd ��|d � d �d }|| jvr�i | j|< || j| |d < q| jS )aW Binary package information available for this release. Reformat the USN.release_packages response to key it based on source package name and related binary package names. :return: Dict keyed by source package name. The second-level key will be binary package names generated from that source package and the values will be the dict response from USN.release_packages for that binary package. The binary metadata contains the following keys: name, version. Optional additional keys: pocket and component. �_release_packages�release_packages� is_sourcer� �sourcez6{usn} metadata defines duplicate source packages {pkg})�usnr ru �� error_msg�issue� extra_info� source_linkzL{issue} metadata does not define release_packages source_link for {bin_pkg}.)r� �bin_pkg�/zX{issue} metadata has unexpected release_packages source_link value for {bin_pkg}: {link})r� r� �link���) r� r� r r� r� r_ r> r �SecurityAPIMetadataErrorrk rt �split)r r� r �source_pkg_namer r r r� x sR �� �� �� zUSN.release_packagesN)r# r$ r% r0 r6 r r' r r! r� r� r5 rt r rs rU ro r� r� r� r r r r rr J s ,rr ri �installed_packagesr: c C s8 i }| j �� D ]\}}|jdkrq||v r|||< q|S )z�Get a dict of any CVEPackageStatuses affecting this Ubuntu release. :return: Dict of active CVEPackageStatus keyed by source package names. r* )r� �itemsrT )ri r� �affected_pkg_versionsr �package_statusr r r �'get_cve_affected_source_packages_status� s �r� c C s| d} t �dd| d dg�\}}i }|�� D ]%}|�d�\}}}}|s%|}d|vr*q||v r5||| |< q||i||< q|S )z�Return a dict of all source packages installed on the system. The dict keys will be source package name: "krb5". The value will be a dict with keys binary_pkg and version. z${db:Status-Status}z dpkg-queryz#-f=${Package},${Source},${Version},� z-W�,� installed)r �subp� splitlinesr� ) � status_field�out�_errr� �pkg_line�pkg_namer� � pkg_versionrT r r r �#query_installed_source_pkg_versions� s$ �� r� c C sn | j sg S i }| j D ] }|jD ]}|�d�sq|| jkr q||vr)|j|d�||< qq tt|�� dd� d��S )z�For a give usn, get the related USNs for it. For each CVE associated with the given USN, we capture other USNs that are related to the CVE. We consider those USNs related to the original USN. rv )r� c S r{ r r| r} r r r r r� z"get_related_usns.<locals>.<lambda>r� )ro r� rw rt r� �listr� �values)r� r^ � related_usnsri �related_usn_idr r r �get_related_usns� s ��� r c C s` i }| D ])}t ||��� D ] \}}||vr|||< q || j}t�||j�dkr,|||< q q|S )Nr )r� r� r r �version_compare)ro r� � affected_pkgsri r� r� � current_verr r r � get_affected_packages_from_cves s" � � ��� r c C s� i }| j �� D ]6\}}||vrqtt�}d|d< dd� |�� D �}|s0tjd�| j�| jdd��|�� |d< t |d �||< q|S ) Nr� rT c S s" h | ] \}}|� d �r|d �qS )r� �r> )r` �_� pkg_bin_infor r r � <setcomp>' s ��z1get_affected_packages_from_usn.<locals>.<setcomp>zC{} metadata defines no pocket information for any release packages.ru r� r� )r� ) r� r� r r' r r� rk rt �popr� )r� r� r r� �pkg_infor� � all_pocketsr r r � get_affected_packages_from_usn s& �� r r� c C s | j r t| j |�S t| |�S )z�Walk CVEs related to a USN and return a dict of all affected packages. :return: Dict keyed on source package name, with active CVEPackageStatus for the current Ubuntu release. )ro r r )r� r� r r r � get_usn_affected_packages_status< s r r� �usn_src_released_pkgsc C sh t �| �}|r2|�d�r2d|jd< |d d |jd< |�� D ]\}}|�d�}|r1||jd< |S q |S )a� Parse release status based on both pkg_status and USN.release_packages. Since some source packages in universe are not represented in CVEPackageStatus, rely on presence of such source packages in usn_src_released_pkgs to represent package as a "released" status. :param pkg_status: the CVEPackageStatus for this source package. :param usn_src_released_pkgs: The USN.release_packages representing only this source package. Normally, release_packages would have data on multiple source packages. :return: Tuple of: human-readable status message, boolean whether released, boolean whether the fix requires access to UA r� r� rT rS r� r� )�copy�deepcopyr> r_ r� )r� r �usn_pkg_statusr� �usn_released_pkgr� r r r �#override_usn_release_package_statusJ s � � �r c C sd i }t | �� �D ]'\}}|�|i �}t||�}|j�dd�}||vr&g ||< || �||f� q|S )Nr� r� )r� r� r> r rT �replace�append)�affected_pkg_status�usn_released_pkgs� status_groups�src_pkgr� �usn_released_srcr � status_groupr r r �group_by_usn_package_statusm s �r �usns� beta_pocketsc s� i }| D ]R}|j �� D ]J\}}� fdd�|�� D �}||vr%|r%|||< q ||v rU|| }|�� D ]#\}} ||vr>| ||< q1|| d } | d } t�| | �dkrT| ||< q1q q|S )a Walk related USNs, merging the released binary package versions. For each USN, iterate over release_packages to collect released binary package names and required fix version. If multiple related USNs require different version fixes to the same binary package, track the maximum version required across all USNs. :param usns: List of USN response instances from which to calculate merge. :param beta_pockets: Dict keyed on service name: esm-infra, esm-apps the values of which will be true of USN response instances from which to calculate merge. :return: Dict keyed by source package name. Under each source package will be a dict with binary package name as keys and binary package metadata as the value. c s. i | ]\}}d � � |� dd�d �u r||�qS )Fr� �Noner )r` � bin_pkg_name� bin_pkg_md�r r r � <dictcomp>� s ��z>merge_usn_released_binary_package_versions.<locals>.<dictcomp>rS r )r� r� r r ) r r �usn_pkg_versionsr� r �binary_pkg_versions�public_bin_pkg_versions� usn_src_pkgr� � binary_pkg_md� prev_version�current_versionr r# r �*merge_usn_released_binary_package_versions{ s, � ���r, �issue_idc C sf t �� }|d ur1|j d ur1|j jd ur1|j jD ]}|j| �� kr0|jr0|j jp(d}tj|f S qdS )NzN/A)NN) r rT �fixesr� �lower�patchedrS r r1 )r- � lp_status�fixrS r r r � _check_cve_fixed_by_livepatch� s �r3 r� c C s| | dkrt jS | dkrt jS | dkrt jS | dv r t jS | dkr#t jS | dkr*t jS | dkr7|r7t jj|d�S t j j| d �S r� ) r r� r� r� r� r� r� r� rk r� )rT r� r r r r� � s" �r� �current_fix_status� fix_statusc C sB | s|S |t jjjt jjjfv r | t jjjkr | |kr |S | S |S r )r r1 r r r2 )r4 r5 r r r �get_expected_overall_status� s �r6 r )4r �enumr� � collectionsr �typingr r r r r r �uaclientr r r r r r � uaclient.httpr �CVE_OR_USN_REGEXrc rj r� r� �STANDARD_UPDATES_POCKET�ESM_INFRA_POCKET�ESM_APPS_POCKETr' r r �Enumr �UAServiceClientr6 r� rU rr r� r� r r r r r r r� r, r3 r� r6 r r r r �<module> s� ��� �� H>t�� � �� ��� �#� � �2� ���� plan/v1.py 0000644 00000110146 15030137052 0006375 0 ustar 00 import enum import re from collections import defaultdict from datetime import datetime, timezone from typing import Any, Dict, List, NamedTuple, Optional, Tuple from uaclient import apt, exceptions, messages from uaclient.api.u.pro.security.fix._common import ( CVE, CVE_OR_USN_REGEX, USN, BinaryPackageFix, CVEPackageStatus, FixStatus, UASecurityClient, _check_cve_fixed_by_livepatch, get_affected_packages_from_usn, get_cve_affected_source_packages_status, get_related_usns, group_by_usn_package_status, merge_usn_released_binary_package_versions, query_installed_source_pkg_versions, ) from uaclient.api.u.pro.status.enabled_services.v1 import _enabled_services from uaclient.api.u.pro.status.is_attached.v1 import ( ContractExpiryStatus, _is_attached, ) from uaclient.config import UAConfig from uaclient.data_types import ( DataObject, Field, IntDataValue, StringDataValue, data_list, ) STANDARD_UPDATES_POCKET = "standard-updates" ESM_INFRA_POCKET = "esm-infra" ESM_APPS_POCKET = "esm-apps" UnfixedPackage = NamedTuple( "UnfixedPackage", [ ("source_package", str), ("binary_package", str), ("version", Optional[str]), ], ) @enum.unique class FixStepType(enum.Enum): ATTACH = "attach" ENABLE = "enable" NOOP = "no-op" APT_UPGRADE = "apt-upgrade" @enum.unique class FixPlanNoOpStatus(enum.Enum): ALREADY_FIXED = "cve-already-fixed" NOT_AFFECTED = "system-not-affected" FIXED_BY_LIVEPATCH = "cve-fixed-by-livepatch" @enum.unique class FixPlanAttachReason(enum.Enum): EXPIRED_CONTRACT = "expired-contract-token" REQUIRED_PRO_SERVICE = "required-pro-service" @enum.unique class FixWarningType(enum.Enum): PACKAGE_CANNOT_BE_INSTALLED = "package-cannot-be-installed" SECURITY_ISSUE_NOT_FIXED = "security-issue-not-fixed" FAIL_UPDATING_ESM_CACHE = "fail-updating-esm-cache" class FixPlanStep(DataObject): fields = [ Field( "operation", StringDataValue, doc=( "The operation that would be performed to fix the issue. This" " can be either an attach, enable, apt-upgrade or a no-op type" ), ), Field( "order", IntDataValue, doc="The execution order of the operation" ), Field( "data", DataObject, doc=( "A data object that can be either an ``AptUpgradeData``," " ``AttachData``, ``EnableData``, ``NoOpData``" ), ), ] def __init__(self, *, operation: str, order: int): self.operation = operation self.order = order class AptUpgradeData(DataObject): fields = [ Field( "binary_packages", data_list(StringDataValue), doc="A list of binary packages that need to be upgraded", ), Field( "source_packages", data_list(StringDataValue), doc="A list of source packages that need to be upgraded", ), Field( "pocket", StringDataValue, doc="The pocket where the packages will be installed from", ), ] def __init__( self, *, binary_packages: List[str], source_packages: List[str], pocket: str ): self.binary_packages = binary_packages self.source_packages = source_packages self.pocket = pocket class FixPlanAptUpgradeStep(FixPlanStep): fields = [ Field("operation", StringDataValue), Field("data", AptUpgradeData), Field("order", IntDataValue), ] def __init__(self, *, data: AptUpgradeData, order: int): super().__init__(operation=FixStepType.APT_UPGRADE.value, order=order) self.data = data class AttachData(DataObject): fields = [ Field( "reason", StringDataValue, doc="The reason why an attach operation is needed", ), Field( "required_service", StringDataValue, doc="The required service that requires the attach operation", ), Field( "source_packages", data_list(StringDataValue), doc="The source packages that require the attach operation", ), ] def __init__( self, *, reason: str, source_packages: List[str], required_service: str ): self.reason = reason self.source_packages = source_packages self.required_service = required_service class FixPlanAttachStep(FixPlanStep): fields = [ Field("operation", StringDataValue), Field("data", AttachData), Field("order", IntDataValue), ] def __init__(self, *, data: AttachData, order: int): super().__init__(operation=FixStepType.ATTACH.value, order=order) self.data = data class EnableData(DataObject): fields = [ Field( "service", StringDataValue, doc="The service that needs to be enabled", ), Field( "source_packages", data_list(StringDataValue), doc="The source packages that require the service to be enabled", ), ] def __init__(self, *, service: str, source_packages: List[str]): self.service = service self.source_packages = source_packages class FixPlanEnableStep(FixPlanStep): fields = [ Field("operation", StringDataValue), Field("data", EnableData), Field("order", IntDataValue), ] def __init__(self, *, data: EnableData, order: int): super().__init__(operation=FixStepType.ENABLE.value, order=order) self.data = data class NoOpData(DataObject): fields = [ Field( "status", StringDataValue, doc="The status of the issue when no operation can be performed", ), ] def __init__(self, *, status: str): self.status = status class FixPlanNoOpStep(FixPlanStep): fields = [ Field("operation", StringDataValue), Field("data", NoOpData), Field("order", IntDataValue), ] def __init__(self, *, data: NoOpData, order: int): super().__init__(operation=FixStepType.NOOP.value, order=order) self.data = data class NoOpLivepatchFixData(NoOpData): fields = [ Field( "status", StringDataValue, doc="The status of the CVE when no operation can be performed", ), Field( "patch_version", StringDataValue, doc="Version of the patch from Livepatch that fixed the CVE", ), ] def __init__(self, *, status: str, patch_version: str): super().__init__(status=status) self.patch_version = patch_version class FixPlanNoOpLivepatchFixStep(FixPlanNoOpStep): fields = [ Field("operation", StringDataValue), Field("data", NoOpLivepatchFixData), Field("order", IntDataValue), ] def __init__(self, *, data: NoOpLivepatchFixData, order: int): super().__init__(data=data, order=order) class NoOpAlreadyFixedData(NoOpData): fields = [ Field( "status", StringDataValue, doc="The status of the issue when no operation can be performed", ), Field( "source_packages", data_list(StringDataValue), doc="The source packages that are already fixed", ), Field( "pocket", StringDataValue, doc="The pocket where the packages would have been installed from", ), ] def __init__( self, *, status: str, source_packages: List[str], pocket: str ): super().__init__(status=status) self.source_packages = source_packages self.pocket = pocket class FixPlanNoOpAlreadyFixedStep(FixPlanNoOpStep): fields = [ Field("operation", StringDataValue), Field("data", NoOpLivepatchFixData), Field("order", IntDataValue), ] def __init__(self, *, data: NoOpAlreadyFixedData, order: int): super().__init__(data=data, order=order) class FixPlanWarning(DataObject): fields = [ Field("warning_type", StringDataValue, doc="The type of warning"), Field( "order", IntDataValue, doc="The execution order of the operation" ), Field( "data", DataObject, doc="A data object that represents either a" " ``PackageCannotBeInstalledData`` or a" " ``SecurityIssueNotFixedData``", ), ] def __init__(self, *, warning_type: str, order: int): self.warning_type = warning_type self.order = order class SecurityIssueNotFixedData(DataObject): fields = [ Field( "source_packages", data_list(StringDataValue), doc="A list of source packages that cannot be fixed at the moment", ), Field( "status", StringDataValue, doc="The status of the CVE regarding those packages", ), ] def __init__(self, *, source_packages: List[str], status: str): self.source_packages = source_packages self.status = status class FixPlanWarningSecurityIssueNotFixed(FixPlanWarning): fields = [ Field("warning_type", StringDataValue), Field("order", IntDataValue), Field("data", SecurityIssueNotFixedData), ] def __init__(self, *, order: int, data: SecurityIssueNotFixedData): super().__init__( warning_type=FixWarningType.SECURITY_ISSUE_NOT_FIXED.value, order=order, ) self.data = data class PackageCannotBeInstalledData(DataObject): fields = [ Field( "binary_package", StringDataValue, doc="The binary package that cannot be installed", ), Field( "binary_package_version", StringDataValue, doc="The version of the binary package that cannot be installed", ), Field( "source_package", StringDataValue, doc="The source package associated with the binary package", ), Field( "related_source_packages", data_list(StringDataValue), doc=( "A list of source packages that come from the same pocket as" " the affected package" ), ), Field( "pocket", StringDataValue, doc=( "The pocket where the affected package should be installed" " from" ), ), ] def __init__( self, *, binary_package: str, binary_package_version: str, source_package: str, pocket: str, related_source_packages: List[str] ): self.source_package = source_package self.binary_package = binary_package self.binary_package_version = binary_package_version self.pocket = pocket self.related_source_packages = related_source_packages class FixPlanWarningPackageCannotBeInstalled(FixPlanWarning): fields = [ Field("warning_type", StringDataValue), Field("order", IntDataValue), Field("data", SecurityIssueNotFixedData), ] def __init__(self, *, order: int, data: PackageCannotBeInstalledData): super().__init__( warning_type=FixWarningType.PACKAGE_CANNOT_BE_INSTALLED.value, order=order, ) self.data = data class FailUpdatingESMCacheData(DataObject): fields = [ Field("title", StringDataValue), Field("code", StringDataValue), ] def __init__(self, *, title: str, code: str): self.title = title self.code = code class FixPlanWarningFailUpdatingESMCache(FixPlanWarning): fields = [ Field("warning_type", StringDataValue), Field("order", IntDataValue), Field("data", FailUpdatingESMCacheData), ] def __init__(self, *, order: int, data: FailUpdatingESMCacheData): super().__init__( warning_type=FixWarningType.FAIL_UPDATING_ESM_CACHE.value, order=order, ) self.data = data class FixPlanError(DataObject): fields = [ Field("msg", StringDataValue, doc="The error message"), Field("code", StringDataValue, required=False, doc="The message code"), ] def __init__(self, *, msg: str, code: Optional[str]): self.msg = msg self.code = code class AdditionalData(DataObject): pass class USNAdditionalData(AdditionalData): fields = [ Field( "associated_cves", data_list(StringDataValue), doc="The associated CVEs for the USN", ), Field( "associated_launchpad_bugs", data_list(StringDataValue), doc="The associated Launchpad bugs for the USN", ), ] def __init__( self, *, associated_cves: List[str], associated_launchpad_bugs: List[str] ): self.associated_cves = associated_cves self.associated_launchpad_bugs = associated_launchpad_bugs class FixPlanResult(DataObject): fields = [ Field("title", StringDataValue, doc="The title of the issue"), Field( "description", StringDataValue, required=False, doc="The description of the issue", ), Field( "current_status", StringDataValue, required=False, doc="The current status of the issue on the system", ), Field( "expected_status", StringDataValue, doc="The expected status of fixing the issue", ), Field( "affected_packages", data_list(StringDataValue), required=False, doc="A list of package names affected by the issue", ), Field( "plan", data_list(FixPlanStep), doc="A list of ``FixPlanStep`` objects", ), Field( "warnings", data_list(FixPlanWarning), required=False, doc="A list of ``FixPlanWarning`` objects", ), Field( "error", FixPlanError, required=False, doc="A ``FixPlanError`` object, if an error occurred.", ), Field( "additional_data", AdditionalData, required=False, doc="Additional data for the issue", ), ] def __init__( self, *, title: str, expected_status: str, plan: List[FixPlanStep], warnings: List[FixPlanWarning], error: Optional[FixPlanError], additional_data: AdditionalData, description: Optional[str] = None, affected_packages: Optional[List[str]] = None, current_status: Optional[str] = None ): self.title = title self.description = description self.current_status = current_status self.expected_status = expected_status self.affected_packages = affected_packages self.plan = plan self.warnings = warnings self.error = error self.additional_data = additional_data class FixPlanUSNResult(DataObject): fields = [ Field( "target_usn_plan", FixPlanResult, doc="A ``FixPlanResult`` object for the target USN", ), Field( "related_usns_plan", data_list(FixPlanResult), required=False, doc="A list of ``FixPlanResult`` objects for the related USNs", ), ] def __init__( self, *, target_usn_plan: FixPlanResult, related_usns_plan: List[FixPlanResult] ): self.target_usn_plan = target_usn_plan self.related_usns_plan = related_usns_plan class FixPlan: def __init__( self, title: str, description: Optional[str], affected_packages: Optional[List[str]] = None, current_status: Optional[str] = None, ): self.order = 1 self.title = title self.description = description self.affected_packages = affected_packages self.current_status = current_status self.fix_steps = [] # type: List[FixPlanStep] self.fix_warnings = [] # type: List[FixPlanWarning] self.error = None # type: Optional[FixPlanError] self.additional_data = AdditionalData() def register_step( self, operation: FixStepType, data: Dict[str, Any], ): # just to make mypy happy fix_step = None # type: Optional[FixPlanStep] if operation == FixStepType.ATTACH: fix_step = FixPlanAttachStep( order=self.order, data=AttachData.from_dict(data) ) elif operation == FixStepType.ENABLE: fix_step = FixPlanEnableStep( order=self.order, data=EnableData.from_dict(data) ) elif operation == FixStepType.NOOP: if "patch_version" in data: fix_step = FixPlanNoOpLivepatchFixStep( order=self.order, data=NoOpLivepatchFixData.from_dict(data) ) elif "source_packages" in data: fix_step = FixPlanNoOpAlreadyFixedStep( order=self.order, data=NoOpAlreadyFixedData.from_dict(data) ) else: fix_step = FixPlanNoOpStep( order=self.order, data=NoOpData.from_dict(data) ) else: fix_step = FixPlanAptUpgradeStep( order=self.order, data=AptUpgradeData.from_dict(data) ) self.fix_steps.append(fix_step) self.order += 1 def register_warning( self, warning_type: FixWarningType, data: Dict[str, Any] ): fix_warning = None # type: Optional[FixPlanWarning] if warning_type == FixWarningType.SECURITY_ISSUE_NOT_FIXED: fix_warning = FixPlanWarningSecurityIssueNotFixed( order=self.order, data=SecurityIssueNotFixedData.from_dict(data), ) elif warning_type == FixWarningType.PACKAGE_CANNOT_BE_INSTALLED: fix_warning = FixPlanWarningPackageCannotBeInstalled( order=self.order, data=PackageCannotBeInstalledData.from_dict(data), ) else: fix_warning = FixPlanWarningFailUpdatingESMCache( order=self.order, data=FailUpdatingESMCacheData.from_dict(data), ) self.fix_warnings.append(fix_warning) self.order += 1 def register_error(self, error_msg: str, error_code: Optional[str]): self.error = FixPlanError(msg=error_msg, code=error_code) def register_additional_data(self, additional_data: Dict[str, Any]): self.additional_data = AdditionalData(**additional_data) def _get_expected_status(self) -> str: if self.error: return "error" if ( len(self.fix_steps) == 1 and isinstance(self.fix_steps[0], FixPlanNoOpStep) and self.fix_steps[0].data.status == "system-not-affected" ): return FixStatus.SYSTEM_NOT_AFFECTED.value.msg elif self.fix_warnings: return FixStatus.SYSTEM_STILL_VULNERABLE.value.msg else: return FixStatus.SYSTEM_NON_VULNERABLE.value.msg @property def fix_plan(self): return FixPlanResult( title=self.title, description=self.description, expected_status=self._get_expected_status(), current_status=self.current_status, affected_packages=self.affected_packages, plan=self.fix_steps, warnings=self.fix_warnings, error=self.error, additional_data=self.additional_data, ) class USNFixPlan(FixPlan): def register_additional_data(self, additional_data: Dict[str, Any]): self.additional_data = USNAdditionalData(**additional_data) def get_fix_plan( title: str, description: Optional[str] = None, affected_packages: Optional[List[str]] = None, ): if not title or "cve" in title.lower(): return FixPlan( title=title, description=description, affected_packages=affected_packages, ) return USNFixPlan( title=title, description=description, affected_packages=affected_packages, ) def _get_cve_data( issue_id: str, client: UASecurityClient, ) -> Tuple[CVE, List[USN]]: try: cve = client.get_cve(cve_id=issue_id) usns = client.get_notices(cves=issue_id) except exceptions.SecurityAPIError as e: if e.code == 404: raise exceptions.SecurityIssueNotFound(issue_id=issue_id) raise e return cve, usns def _get_usn_data( issue_id: str, client: UASecurityClient ) -> Tuple[USN, List[USN]]: try: usn = client.get_notice(notice_id=issue_id) usns = get_related_usns(usn, client) except exceptions.SecurityAPIError as e: if e.code == 404: raise exceptions.SecurityIssueNotFound(issue_id=issue_id) raise e if not usn.response["release_packages"]: # Since usn.release_packages filters to our current release only # check overall metadata and error if empty. raise exceptions.SecurityAPIMetadataError( error_msg="metadata defines no fixed package versions.", issue=issue_id, extra_info="", ) return usn, usns def _get_upgradable_pkgs( binary_pkgs: List[BinaryPackageFix], check_esm_cache: bool, ) -> Tuple[List[str], List[UnfixedPackage]]: upgrade_pkgs = [] unfixed_pkgs = [] for binary_pkg in sorted(binary_pkgs): candidate_version = apt.get_pkg_candidate_version( binary_pkg.binary_pkg, check_esm_cache=check_esm_cache ) if ( candidate_version and apt.version_compare( binary_pkg.fixed_version, candidate_version ) <= 0 ): upgrade_pkgs.append(binary_pkg.binary_pkg) else: unfixed_pkgs.append( UnfixedPackage( source_package=binary_pkg.source_pkg, binary_package=binary_pkg.binary_pkg, version=binary_pkg.fixed_version, ) ) return upgrade_pkgs, unfixed_pkgs def _get_upgradable_package_candidates_by_pocket( pkg_status_group: List[Tuple[str, CVEPackageStatus]], usn_released_pkgs: Dict[str, Dict[str, Dict[str, str]]], installed_pkgs: Dict[str, Dict[str, str]], ): binary_pocket_pkgs = defaultdict(list) src_pocket_pkgs = defaultdict(list) for src_pkg, pkg_status in pkg_status_group: src_pocket_pkgs[pkg_status.pocket_source].append((src_pkg, pkg_status)) for binary_pkg, version in installed_pkgs[src_pkg].items(): usn_released_src = usn_released_pkgs.get(src_pkg, {}) if binary_pkg not in usn_released_src: continue fixed_version = usn_released_src.get(binary_pkg, {}).get( "version", "" ) if apt.version_compare(fixed_version, version) > 0: binary_pocket_pkgs[pkg_status.pocket_source].append( BinaryPackageFix( source_pkg=src_pkg, binary_pkg=binary_pkg, fixed_version=fixed_version, ) ) return src_pocket_pkgs, binary_pocket_pkgs def _get_cve_description( cve: CVE, installed_pkgs: Dict[str, Dict[str, str]], ): if not cve.notices: return cve.description for notice in cve.notices: usn_pkgs = notice.release_packages.keys() for pkg in usn_pkgs: if pkg in installed_pkgs: return notice.title return cve.notices[0].title def _fix_plan_cve(issue_id: str, cfg: UAConfig) -> FixPlanResult: livepatch_cve_status, patch_version = _check_cve_fixed_by_livepatch( issue_id ) if livepatch_cve_status: fix_plan = get_fix_plan(title=issue_id) fix_plan.current_status = FixStatus.SYSTEM_NON_VULNERABLE.value.msg fix_plan.register_step( operation=FixStepType.NOOP, data={ "status": FixPlanNoOpStatus.FIXED_BY_LIVEPATCH.value, "patch_version": patch_version, }, ) return fix_plan.fix_plan client = UASecurityClient(cfg=cfg) installed_pkgs = query_installed_source_pkg_versions() try: cve, usns = _get_cve_data(issue_id=issue_id, client=client) except ( exceptions.SecurityIssueNotFound, exceptions.SecurityAPIError, ) as e: fix_plan = get_fix_plan(title=issue_id) fix_plan.register_error(error_msg=e.msg, error_code=e.msg_code) return fix_plan.fix_plan affected_pkg_status = get_cve_affected_source_packages_status( cve=cve, installed_packages=installed_pkgs ) usn_released_pkgs = merge_usn_released_binary_package_versions( usns, beta_pockets={} ) cve_description = _get_cve_description(cve, installed_pkgs) return _generate_fix_plan( issue_id=issue_id, issue_description=cve_description, affected_pkg_status=affected_pkg_status, usn_released_pkgs=usn_released_pkgs, installed_pkgs=installed_pkgs, cfg=cfg, ) def _fix_plan_usn(issue_id: str, cfg: UAConfig) -> FixPlanUSNResult: client = UASecurityClient(cfg=cfg) installed_pkgs = query_installed_source_pkg_versions() try: usn, related_usns = _get_usn_data(issue_id=issue_id, client=client) except ( exceptions.SecurityIssueNotFound, exceptions.SecurityAPIError, ) as e: fix_plan = get_fix_plan(title=issue_id) fix_plan.register_error(error_msg=e.msg, error_code=e.msg_code) return FixPlanUSNResult( target_usn_plan=fix_plan.fix_plan, related_usns_plan=[], ) affected_pkg_status = get_affected_packages_from_usn( usn=usn, installed_packages=installed_pkgs ) usn_released_pkgs = merge_usn_released_binary_package_versions( [usn], beta_pockets={} ) additional_data = { "associated_cves": [] if not usn.cves_ids else usn.cves_ids, "associated_launchpad_bugs": ( [] if not usn.references else usn.references ), } target_usn_plan = _generate_fix_plan( issue_id=issue_id, issue_description=usn.title, affected_pkg_status=affected_pkg_status, usn_released_pkgs=usn_released_pkgs, installed_pkgs=installed_pkgs, cfg=cfg, additional_data=additional_data, ) related_usns_plan = [] # type: List[FixPlanResult] for usn in related_usns: affected_pkg_status = get_affected_packages_from_usn( usn=usn, installed_packages=installed_pkgs ) usn_released_pkgs = merge_usn_released_binary_package_versions( [usn], beta_pockets={} ) additional_data = { "associated_cves": [] if not usn.cves_ids else usn.cves_ids, "associated_launchpad_bugs": ( [] if not usn.references else usn.references ), } related_usns_plan.append( _generate_fix_plan( issue_id=usn.id, issue_description=usn.title, affected_pkg_status=affected_pkg_status, usn_released_pkgs=usn_released_pkgs, installed_pkgs=installed_pkgs, cfg=cfg, additional_data=additional_data, ) ) return FixPlanUSNResult( target_usn_plan=target_usn_plan, related_usns_plan=related_usns_plan, ) def fix_plan_cve(issue_id: str, cfg: UAConfig) -> FixPlanResult: if not issue_id or not re.match(CVE_OR_USN_REGEX, issue_id): fix_plan = get_fix_plan(title=issue_id) msg = messages.INVALID_SECURITY_ISSUE.format(issue_id=issue_id) fix_plan.register_error(error_msg=msg.msg, error_code=msg.name) return fix_plan.fix_plan issue_id = issue_id.upper() return _fix_plan_cve(issue_id, cfg) def fix_plan_usn(issue_id: str, cfg: UAConfig) -> FixPlanUSNResult: if not issue_id or not re.match(CVE_OR_USN_REGEX, issue_id): fix_plan = get_fix_plan(title=issue_id) msg = messages.INVALID_SECURITY_ISSUE.format(issue_id=issue_id) fix_plan.register_error(error_msg=msg.msg, error_code=msg.name) return FixPlanUSNResult( target_usn_plan=fix_plan.fix_plan, related_usns_plan=[], ) issue_id = issue_id.upper() return _fix_plan_usn(issue_id, cfg) def get_pocket_short_name(pocket: str): if pocket == messages.SECURITY_UBUNTU_STANDARD_UPDATES_POCKET: return STANDARD_UPDATES_POCKET elif pocket == messages.SECURITY_UA_INFRA_POCKET: return ESM_INFRA_POCKET elif pocket == messages.SECURITY_UA_APPS_POCKET: return ESM_APPS_POCKET else: return pocket def _should_update_esm_cache( check_esm_cache: bool, esm_cache_updated: bool, cfg: UAConfig, ) -> bool: if ( check_esm_cache and not esm_cache_updated and not _is_attached(cfg).is_attached ): last_apt_update = apt.get_apt_cache_datetime() if last_apt_update is None: return True now = datetime.now(timezone.utc) time_since_update = now - last_apt_update if time_since_update.days > 2: return True return False def _generate_fix_plan( *, issue_id: str, issue_description: str, affected_pkg_status: Dict[str, CVEPackageStatus], usn_released_pkgs: Dict[str, Dict[str, Dict[str, str]]], installed_pkgs: Dict[str, Dict[str, str]], cfg: UAConfig, additional_data=None ) -> FixPlanResult: count = len(affected_pkg_status) src_pocket_pkgs = defaultdict(list) esm_cache_updated = False fix_plan = get_fix_plan( title=issue_id, description=issue_description, affected_packages=sorted(list(affected_pkg_status.keys())), ) if additional_data: fix_plan.register_additional_data(additional_data) if count == 0: fix_plan.current_status = FixStatus.SYSTEM_NOT_AFFECTED.value.msg fix_plan.register_step( operation=FixStepType.NOOP, data={"status": FixPlanNoOpStatus.NOT_AFFECTED.value}, ) return fix_plan.fix_plan else: fix_plan.current_status = FixStatus.SYSTEM_STILL_VULNERABLE.value.msg pkg_status_groups = group_by_usn_package_status( affected_pkg_status, usn_released_pkgs ) for status_value, pkg_status_group in sorted(pkg_status_groups.items()): if status_value != "released": fix_plan.register_warning( warning_type=FixWarningType.SECURITY_ISSUE_NOT_FIXED, data={ "source_packages": [ src_pkg for src_pkg, _ in pkg_status_group ], "status": status_value, }, ) fix_plan.current_status = ( FixStatus.SYSTEM_STILL_VULNERABLE.value.msg ) else: ( src_pocket_pkgs, binary_pocket_pkgs, ) = _get_upgradable_package_candidates_by_pocket( pkg_status_group, usn_released_pkgs, installed_pkgs, ) if not src_pocket_pkgs: return fix_plan.fix_plan for pocket in [ messages.SECURITY_UBUNTU_STANDARD_UPDATES_POCKET, messages.SECURITY_UA_INFRA_POCKET, messages.SECURITY_UA_APPS_POCKET, ]: pkg_src_group = src_pocket_pkgs[pocket] binary_pkgs = binary_pocket_pkgs[pocket] source_pkgs = [src_pkg for src_pkg, _ in pkg_src_group] pocket_name = get_pocket_short_name(pocket) if not binary_pkgs: if source_pkgs: fix_plan.current_status = ( FixStatus.SYSTEM_NON_VULNERABLE.value.msg ) fix_plan.register_step( operation=FixStepType.NOOP, data={ "status": FixPlanNoOpStatus.ALREADY_FIXED.value, "source_packages": source_pkgs, "pocket": pocket_name, }, ) continue check_esm_cache = ( pocket != messages.SECURITY_UBUNTU_STANDARD_UPDATES_POCKET ) if _should_update_esm_cache(check_esm_cache, esm_cache_updated, cfg): try: apt.update_esm_caches(cfg) esm_cache_updated = True except Exception as e: error_msg = messages.E_UPDATING_ESM_CACHE.format( error=getattr(e, "msg", str(e)) ) fix_plan.register_warning( warning_type=FixWarningType.FAIL_UPDATING_ESM_CACHE, data={ "title": error_msg.msg, "code": error_msg.name, }, ) upgrade_pkgs, unfixed_pkgs = _get_upgradable_pkgs( binary_pkgs, check_esm_cache ) if unfixed_pkgs: fix_plan.current_status = ( FixStatus.SYSTEM_STILL_VULNERABLE.value.msg ) for unfixed_pkg in unfixed_pkgs: fix_plan.register_warning( warning_type=FixWarningType.PACKAGE_CANNOT_BE_INSTALLED, data={ "binary_package": unfixed_pkg.binary_package, "binary_package_version": unfixed_pkg.version, "source_package": unfixed_pkg.source_package, "related_source_packages": source_pkgs, "pocket": pocket_name, }, ) if pocket != messages.SECURITY_UBUNTU_STANDARD_UPDATES_POCKET: if pocket == messages.SECURITY_UA_INFRA_POCKET: service_to_check = "esm-infra" else: service_to_check = "esm-apps" if not _is_attached(cfg).is_attached: fix_plan.register_step( operation=FixStepType.ATTACH, data={ "reason": "required-pro-service", "source_packages": source_pkgs, "required_service": service_to_check, }, ) else: contract_expiry_status = _is_attached(cfg).contract_status if contract_expiry_status != ContractExpiryStatus.ACTIVE.value: fix_plan.register_step( operation=FixStepType.ATTACH, data={ "reason": FixPlanAttachReason.EXPIRED_CONTRACT.value, # noqa "source_packages": source_pkgs, }, ) enabled_services = _enabled_services(cfg).enabled_services or [] enabled_services_names = ( [service.name for service in enabled_services] if enabled_services else [] ) if service_to_check not in enabled_services_names: fix_plan.register_step( operation=FixStepType.ENABLE, data={ "service": service_to_check, "source_packages": source_pkgs, }, ) fix_plan.register_step( operation=FixStepType.APT_UPGRADE, data={ "binary_packages": upgrade_pkgs, "source_packages": source_pkgs, "pocket": pocket_name, }, ) return fix_plan.fix_plan plan/__pycache__/__init__.cpython-310.pyc 0000644 00000000264 15030137052 0014144 0 ustar 00 o �ϴf � @ s d S )N� r r r �W/usr/lib/python3/dist-packages/uaclient/api/u/pro/security/fix/_common/plan/__init__.py�<module> s plan/__pycache__/v1.cpython-310.pyc 0000644 00000062566 15030137052 0012750 0 ustar 00 o �P�ff� � @ s d dl Z d dlZd dlmZ d dlmZmZ d dlmZmZm Z m Z m Z m Z d dl mZmZmZ d dlmZmZmZmZmZmZmZmZmZmZm Z m Z m Z m Z d dl m!Z! d dl"m#Z#m$Z$ d d l%m&Z& d d l'm(Z(m)Z)m*Z*m+Z+m,Z, d Z-d Z.d Z/e dde0fde0fde e0 fg�Z1e j2G dd� de j3��Z4e j2G dd� de j3��Z5e j2G dd� de j3��Z6e j2G dd� de j3��Z7G dd� de(�Z8G d d � d e(�Z9G d d � d e8�Z:G d d!� d!e(�Z;G d"d#� d#e8�Z<G d$d%� d%e(�Z=G d&d'� d'e8�Z>G d(d)� d)e(�Z?G d*d+� d+e8�Z@G d,d-� d-e?�ZAG d.d/� d/e@�ZBG d0d1� d1e?�ZCG d2d3� d3e@�ZDG d4d5� d5e(�ZEG d6d7� d7e(�ZFG d8d9� d9eE�ZGG d:d;� d;e(�ZHG d<d=� d=eE�ZIG d>d?� d?e(�ZJG d@dA� dAeE�ZKG dBdC� dCe(�ZLG dDdE� dEe(�ZMG dFdG� dGeM�ZNG dHdI� dIe(�ZOG dJdK� dKe(�ZPG dLdM� dM�ZQG dNdO� dOeQ�ZR d|dPe0dQe e0 dRe e e0 fdSdT�ZSdUe0dVedWe ee e f fdXdY�ZTdUe0dVedWe ee e f fdZd[�ZUd\e e d]eVdWe e e0 e e1 f fd^d_�ZWd`e e e0ef daee0ee0ee0e0f f f dbee0ee0e0f f fdcdd�ZXdeedbee0ee0e0f f fdfdg�ZYdUe0dhe&dWeOfdidj�ZZdUe0dhe&dWePfdkdl�Z[dUe0dhe&dWeOfdmdn�Z\dUe0dhe&dWePfdodp�Z]dqe0fdrds�Z^d]eVdteVdhe&dWeVfdudv�Z_ddw�dUe0dxe0dyee0ef daee0ee0ee0e0f f f dbee0ee0e0f f dhe&dWeOfdzd{�Z`dS )}� N)� defaultdict)�datetime�timezone)�Any�Dict�List� NamedTuple�Optional�Tuple)�apt� exceptions�messages)�CVE�CVE_OR_USN_REGEX�USN�BinaryPackageFix�CVEPackageStatus� FixStatus�UASecurityClient� _check_cve_fixed_by_livepatch� get_affected_packages_from_usn�'get_cve_affected_source_packages_status�get_related_usns�group_by_usn_package_status�*merge_usn_released_binary_package_versions�#query_installed_source_pkg_versions)�_enabled_services)�ContractExpiryStatus� _is_attached)�UAConfig)� DataObject�Field� IntDataValue�StringDataValue� data_listzstandard-updates� esm-infra�esm-apps�UnfixedPackage�source_package�binary_package�versionc @ s e Zd ZdZdZdZdZdS )� FixStepType�attach�enablezno-opz apt-upgradeN)�__name__� __module__� __qualname__�ATTACH�ENABLE�NOOP� APT_UPGRADE� r5 r5 �Q/usr/lib/python3/dist-packages/uaclient/api/u/pro/security/fix/_common/plan/v1.pyr+ 4 s r+ c @ � e Zd ZdZdZdZdS )�FixPlanNoOpStatuszcve-already-fixed�system-not-affectedzcve-fixed-by-livepatchN)r. r/ r0 � ALREADY_FIXED� NOT_AFFECTED�FIXED_BY_LIVEPATCHr5 r5 r5 r6 r8 < � r8 c @ s e Zd ZdZdZdS )�FixPlanAttachReasonzexpired-contract-token�required-pro-serviceN)r. r/ r0 �EXPIRED_CONTRACT�REQUIRED_PRO_SERVICEr5 r5 r5 r6 r> C s r> c @ r7 )�FixWarningTypezpackage-cannot-be-installedzsecurity-issue-not-fixedzfail-updating-esm-cacheN)r. r/ r0 �PACKAGE_CANNOT_BE_INSTALLED�SECURITY_ISSUE_NOT_FIXED�FAIL_UPDATING_ESM_CACHEr5 r5 r5 r6 rB I r= rB c @ �F e Zd Zededd�ededd�ededd�gZdede fdd �Z d S ) � FixPlanStep� operationzyThe operation that would be performed to fix the issue. This can be either an attach, enable, apt-upgrade or a no-op type��doc�order�$The execution order of the operation�datazdA data object that can be either an ``AptUpgradeData``, ``AttachData``, ``EnableData``, ``NoOpData``c C � || _ || _d S �N�rH rK )�selfrH rK r5 r5 r6 �__init__g � zFixPlanStep.__init__N� r. r/ r0 r! r# r" r �fields�str�intrR r5 r5 r5 r6 rG P s ����rG c @ sZ e Zd Zedee�dd�edee�dd�ededd�gZdee dee defdd �Z d S ) �AptUpgradeData�binary_packagesz2A list of binary packages that need to be upgradedrI �source_packagesz2A list of source packages that need to be upgraded�pocketz4The pocket where the packages will be installed fromc C � || _ || _|| _d S rO �rY rZ r[ )rQ rY rZ r[ r5 r5 r6 rR s zAptUpgradeData.__init__N� r. r/ r0 r! r$ r# rU r rV rR r5 r5 r5 r6 rX l s0 �������rX c �B e Zd Zede�ede�ede�gZdedef� fdd� Z � Z S )�FixPlanAptUpgradeSteprH rM rK c � t � jtjj|d� || _d S �NrP )�superrR r+ r4 �valuerM �rQ rM rK �� __class__r5 r6 rR � � z FixPlanAptUpgradeStep.__init__) r. r/ r0 r! r# rX r" rU rW rR � __classcell__r5 r5 rf r6 r` � � � r` c @ sR e Zd Zededd�ededd�edee�dd�gZdedee defdd �Z d S ) � AttachData�reasonz,The reason why an attach operation is neededrI �required_servicez7The required service that requires the attach operationrZ z5The source packages that require the attach operationc C r\ rO �rl rZ rm )rQ rl rZ rm r5 r5 r6 rR � s zAttachData.__init__N� r. r/ r0 r! r# r$ rU rV r rR r5 r5 r5 r6 rk � s0 �������rk c r_ )�FixPlanAttachSteprH rM rK c ra rb )rc rR r+ r1 rd rM re rf r5 r6 rR � rh zFixPlanAttachStep.__init__) r. r/ r0 r! r# rk r" rU rW rR ri r5 r5 rf r6 rp � rj rp c @ sB e Zd Zededd�edee�dd�gZdedee fdd�Z dS ) � EnableData�servicez$The service that needs to be enabledrI rZ z:The source packages that require the service to be enabledc C rN rO �rr rZ )rQ rr rZ r5 r5 r6 rR � rS zEnableData.__init__Nro r5 r5 r5 r6 rq � s ��� rq c r_ )�FixPlanEnableSteprH rM rK c ra rb )rc rR r+ r2 rd rM re rf r5 r6 rR � rh zFixPlanEnableStep.__init__) r. r/ r0 r! r# rq r" rU rW rR ri r5 r5 rf r6 rt � rj rt c @ s* e Zd Zededd�gZdefdd�ZdS )�NoOpData�status�:The status of the issue when no operation can be performedrI c C s || _ d S rO �rv )rQ rv r5 r5 r6 rR � s zNoOpData.__init__N�r. r/ r0 r! r# rU rV rR r5 r5 r5 r6 ru � s ��ru c r_ )�FixPlanNoOpSteprH rM rK c ra rb )rc rR r+ r3 rd rM re rf r5 r6 rR � rh zFixPlanNoOpStep.__init__) r. r/ r0 r! r# ru r" rU rW rR ri r5 r5 rf r6 rz � rj rz c sB e Zd Zededd�ededd�gZdedef� fdd� Z� ZS )�NoOpLivepatchFixDatarv z8The status of the CVE when no operation can be performedrI � patch_versionz6Version of the patch from Livepatch that fixed the CVEc s t � j|d� || _d S �Nrx )rc rR r| )rQ rv r| rf r5 r6 rR s z NoOpLivepatchFixData.__init__) r. r/ r0 r! r# rU rV rR ri r5 r5 rf r6 r{ � s ��� r{ c r_ )�FixPlanNoOpLivepatchFixSteprH rM rK c � t � j||d� d S �N)rM rK �rc rR re rf r5 r6 rR � z$FixPlanNoOpLivepatchFixStep.__init__) r. r/ r0 r! r# r{ r" rU rW rR ri r5 r5 rf r6 r~ rj r~ c sZ e Zd Zededd�edee�dd�ededd�gZdedee def� fdd � Z � Z S ) �NoOpAlreadyFixedDatarv rw rI rZ z*The source packages that are already fixedr[ z<The pocket where the packages would have been installed fromc s t � j|d� || _|| _d S r} )rc rR rZ r[ )rQ rv rZ r[ rf r5 r6 rR ' s z NoOpAlreadyFixedData.__init__) r. r/ r0 r! r# r$ rU rV r rR ri r5 r5 rf r6 r� s0 �������r� c �B e Zd Zede�ede�ede�gZdede f� fdd� Z � Z S )�FixPlanNoOpAlreadyFixedSteprH rM rK c r r� r� re rf r5 r6 rR 6 r� z$FixPlanNoOpAlreadyFixedStep.__init__) r. r/ r0 r! r# r{ r" rU r� rW rR ri r5 r5 rf r6 r� / rj r� c @ rF ) �FixPlanWarning� warning_typezThe type of warningrI rK rL rM zjA data object that represents either a ``PackageCannotBeInstalledData`` or a ``SecurityIssueNotFixedData``c C rN rO �r� rK )rQ r� rK r5 r5 r6 rR I rS zFixPlanWarning.__init__NrT r5 r5 r5 r6 r� : s ���r� c @ sB e Zd Zedee�dd�ededd�gZdee defdd�Z dS ) �SecurityIssueNotFixedDatarZ z<A list of source packages that cannot be fixed at the momentrI rv z.The status of the CVE regarding those packagesc C rN rO �rZ rv )rQ rZ rv r5 r5 r6 rR \ rS z"SecurityIssueNotFixedData.__init__Nr^ r5 r5 r5 r6 r� N s ��� r� c �B e Zd Zede�ede�ede�gZdedef� fdd� Z � Z S )�#FixPlanWarningSecurityIssueNotFixedr� rK rM c � t � jtjj|d� || _d S �Nr� )rc rR rB rD rd rM �rQ rK rM rf r5 r6 rR h � � z,FixPlanWarningSecurityIssueNotFixed.__init__) r. r/ r0 r! r# r" r� rU rW rR ri r5 r5 rf r6 r� a rj r� c @ sr e Zd Zededd�ededd�ededd�edee�d d�ed ed d�gZdededed edee f d d �Z dS )� PackageCannotBeInstalledDatar) z+The binary package that cannot be installedrI �binary_package_versionz:The version of the binary package that cannot be installedr( z5The source package associated with the binary package�related_source_packageszPA list of source packages that come from the same pocket as the affected packager[ z>The pocket where the affected package should be installed fromc C s" || _ || _|| _|| _|| _d S rO )r( r) r� r[ r� )rQ r) r� r( r[ r� r5 r5 r6 rR � s z%PackageCannotBeInstalledData.__init__Nro r5 r5 r5 r6 r� p sL ������"�����r� c r� )�&FixPlanWarningPackageCannotBeInstalledr� rK rM c r� r� )rc rR rB rC rd rM r� rf r5 r6 rR � r� z/FixPlanWarningPackageCannotBeInstalled.__init__) r. r/ r0 r! r# r" r� rU rW r� rR ri r5 r5 rf r6 r� � rj r� c @ s2 e Zd Zede�ede�gZdedefdd�ZdS )�FailUpdatingESMCacheData�title�codec C rN rO �r� r� )rQ r� r� r5 r5 r6 rR � rS z!FailUpdatingESMCacheData.__init__Nry r5 r5 r5 r6 r� � s �r� c r� )�"FixPlanWarningFailUpdatingESMCacher� rK rM c r� r� )rc rR rB rE rd rM r� rf r5 r6 rR � r� z+FixPlanWarningFailUpdatingESMCache.__init__) r. r/ r0 r! r# r" r� rU rW rR ri r5 r5 rf r6 r� � rj r� c @ s@ e Zd Zededd�ededdd�gZdedee fdd �Zd S ) � FixPlanError�msgzThe error messagerI r� FzThe message code��requiredrJ c C rN rO �r� r� )rQ r� r� r5 r5 r6 rR � rS zFixPlanError.__init__N) r. r/ r0 r! r# rU rV r rR r5 r5 r5 r6 r� � s �r� c @ s e Zd ZdS )�AdditionalDataN)r. r/ r0 r5 r5 r5 r6 r� � s r� c @ sJ e Zd Zedee�dd�edee�dd�gZdee dee fdd�Z dS ) �USNAdditionalData�associated_cvesz The associated CVEs for the USNrI �associated_launchpad_bugsz)The associated Launchpad bugs for the USNc C rN rO �r� r� )rQ r� r� r5 r5 r6 rR � � zUSNAdditionalData.__init__Nr^ r5 r5 r5 r6 r� � s" ��� ��r� c @ s� e Zd Zededd�ededdd�ededd d�ed ed d�ed ee�dd d�edee�dd�edee�ddd�ededdd�ede ddd�g Z dddd�de d e de e de e de e de de e d e e e de e fdd�ZdS )� FixPlanResultr� zThe title of the issuerI � descriptionFz The description of the issuer� �current_statusz-The current status of the issue on the system�expected_statusz'The expected status of fixing the issue�affected_packagesz-A list of package names affected by the issue�planz!A list of ``FixPlanStep`` objects�warningsz$A list of ``FixPlanWarning`` objects�errorz0A ``FixPlanError`` object, if an error occurred.�additional_dataz Additional data for the issueN)r� r� r� c C s: || _ || _| | _|| _|| _|| _|| _|| _|| _d S rO ) r� r� r� r� r� r� r� r� r� ) rQ r� r� r� r� r� r� r� r� r� r5 r5 r6 rR ' s zFixPlanResult.__init__)r. r/ r0 r! r# r$ rG r� r� r� rU rV r r rR r5 r5 r5 r6 r� � s� ���������;������� � � �r� c @ sD e Zd Zededd�edee�ddd�gZdedee fdd �Zd S ) �FixPlanUSNResult�target_usn_planz-A ``FixPlanResult`` object for the target USNrI �related_usns_planFz8A list of ``FixPlanResult`` objects for the related USNsr� c C rN rO �r� r� )rQ r� r� r5 r5 r6 rR N r� zFixPlanUSNResult.__init__N) r. r/ r0 r! r� r$ rU r rR r5 r5 r5 r6 r� ? s$ �����r� c @ s� e Zd Z ddedee deee dee fdd�Zded eee f fd d �Z d e d eee f fd d�Z dedee fdd�Z deee f fdd�Zdefdd�Zedd� �ZdS ) �FixPlanNr� r� r� r� c C s< d| _ || _|| _|| _|| _g | _g | _d | _t� | _ d S )N� ) rK r� r� r� r� � fix_steps� fix_warningsr� r� r� )rQ r� r� r� r� r5 r5 r6 rR Y s zFixPlan.__init__rH rM c C s� d }|t jkrt| jt�|�d�}nH|t jkr"t| jt�|�d�}n8|t j krPd|v r6t | jt �|�d�}n$d|v rEt | jt �|�d�}nt| jt�|�d�}n t| jt�|�d�}| j�|� | jd7 _d S )N�rK rM r| rZ r� )r+ r1 rp rK rk � from_dictr2 rt rq r3 r~ r{ r� r� rz ru r` rX r� �append)rQ rH rM �fix_stepr5 r5 r6 � register_stepj s4 � � � � � � zFixPlan.register_stepr� c C sv d }|t jkrt| jt�|�d�}n|t jkr"t| jt�|�d�}n t | jt �|�d�}| j � |� | jd7 _d S )Nr� r� ) rB rD r� rK r� r� rC r� r� r� r� r� r� )rQ r� rM � fix_warningr5 r5 r6 �register_warning� s" � �� zFixPlan.register_warning� error_msg� error_codec C s t ||d�| _d S )Nr� )r� r� )rQ r� r� r5 r5 r6 �register_error� s zFixPlan.register_errorr� c C � t di |��| _d S �Nr5 )r� r� �rQ r� r5 r5 r6 �register_additional_data� r� z FixPlan.register_additional_data�returnc C s^ | j rdS t| j�dkr"t| jd t�r"| jd jjdkr"tjj j S | j r*tj j j S tj j j S )Nr� r� r r9 )r� �lenr� � isinstancerz rM rv r �SYSTEM_NOT_AFFECTEDrd r� r� �SYSTEM_STILL_VULNERABLE�SYSTEM_NON_VULNERABLE�rQ r5 r5 r6 �_get_expected_status� s � z FixPlan._get_expected_statusc C s. t | j| j| �� | j| j| j| j| j| j d� S )N) r� r� r� r� r� r� r� r� r� ) r� r� r� r� r� r� r� r� r� r� r� r5 r5 r6 �fix_plan� s �zFixPlan.fix_plan�NN)r. r/ r0 rV r r rR r+ r r r� rB r� r� r� r� �propertyr� r5 r5 r5 r6 r� X s6 ��� � �� �%� �r� c @ s" e Zd Zdeeef fdd�ZdS )� USNFixPlanr� c C r� r� )r� r� r� r5 r5 r6 r� � r� z#USNFixPlan.register_additional_dataN)r. r/ r0 r rV r r� r5 r5 r5 r6 r� � s r� r� r� r� c C s, | rd| � � v rt| ||d�S t| ||d�S )N�cve�r� r� r� )�lowerr� r� r� r5 r5 r6 � get_fix_plan� s ��r� �issue_id�clientr� c C sX z|j | d�}|j| d�}W ||fS tjy+ } z |jdkr%tj| d��|�d }~ww )N)�cve_id)�cves� �r� )�get_cve� get_noticesr �SecurityAPIErrorr� �SecurityIssueNotFound)r� r� r� �usns�er5 r5 r6 � _get_cve_data� s � ��r� c C sr z |j | d�}t||�}W n tjy' } z |jdkr!tj| d��|�d }~ww |jd s5tjd| dd��||fS )N)� notice_idr� r� �release_packagesz+metadata defines no fixed package versions.� )r� �issue� extra_info)� get_noticer r r� r� r� �response�SecurityAPIMetadataError)r� r� �usnr� r� r5 r5 r6 � _get_usn_data� s �� �r� � binary_pkgs�check_esm_cachec C sl g }g }t | �D ])}tj|j|d�}|r$t�|j|�dkr$|�|j� q|�t|j|j|jd�� q||fS )N)r� r )r( r) r* ) �sortedr �get_pkg_candidate_version� binary_pkg�version_compare� fixed_versionr� r' � source_pkg)r� r� � upgrade_pkgs� unfixed_pkgsr� �candidate_versionr5 r5 r6 �_get_upgradable_pkgs s, ������r� �pkg_status_group�usn_released_pkgs�installed_pkgsc C s� t t�}t t�}| D ]C\}}||j �||f� || �� D ].\}}|�|i �} || vr-q | �|i ��dd�} t�| |�dkrL||j �t||| d�� q q ||fS )Nr* r� r )r� r� r� ) r �list� pocket_sourcer� �items�getr r� r ) r� r� r� �binary_pocket_pkgs�src_pocket_pkgs�src_pkg� pkg_statusr� r* �usn_released_srcr� r5 r5 r6 �,_get_upgradable_package_candidates_by_pocket( s, � ����r r� c C sN | j s| jS | j D ]}|j�� }|D ] }||v r |j S qq | j d jS )Nr )�noticesr� r� �keysr� )r� r� �notice�usn_pkgs�pkgr5 r5 r6 �_get_cve_descriptionF s �� r �cfgc C s� t | �\}}|r#t| d�}tjjj|_|jtj t j j|d�d� |j S t |d�}t� }z t| |d�\}}W n' tjtjfy\ } zt| d�}|j| j| jd� |j W Y d } ~ S d } ~ ww t||d�} t|i d�} t||�} t| | | | ||d �S ) N�r� )rv r| �rH rM �r �r� r� �r� r� )r� �installed_packages�� beta_pockets)r� �issue_description�affected_pkg_statusr� r� r )r r� r r� rd r� r� r� r+ r3 r8 r< r� r r r� r r� r� r� �msg_coder r r �_generate_fix_plan) r� r �livepatch_cve_statusr| r� r� r� r� r� r� r r� �cve_descriptionr5 r5 r6 � _fix_plan_cveV sR � �� � ���� �r c C sF t |d�}t� }z t| |d�\}}W n+ tjtjfy= } zt| d�}|j|j|j d� t |j g d�W Y d }~S d }~ww t ||d�}t |gi d�} |jsPg n|j|jsWg n|jd�} t| |j|| ||| d �} g } |D ]0}t ||d�}t |gi d�} |js�g n|j|js�g n|jd�} | �t|j|j|| ||| d �� qlt | | d�S ) Nr r r r r� )r� r r r� )r� r r r� r� r r� )r r r� r r� r� r� r� r� r r� r� r r �cves_ids� referencesr r� r� �id) r� r r� r� r� � related_usnsr� r� r r� r� r� r� r5 r5 r6 � _fix_plan_usn� sx � ��� ���� ����� �r c C sR | rt �t| �s t| d�}tjj| d�}|j|j|j d� |j S | � � } t | |�S )Nr r� r ) �re�matchr r� r �INVALID_SECURITY_ISSUE�formatr� r� �namer� �upperr �r� r r� r� r5 r5 r6 � fix_plan_cve� s r& c C sZ | rt �t| �s$t| d�}tjj| d�}|j|j|j d� t |j g d�S | � � } t | |�S )Nr r� r r� )r r r r� r r! r" r� r� r# r� r� r$ r r% r5 r5 r6 � fix_plan_usn� s � r' r[ c C s. | t jkrtS | t jkrtS | t jkrtS | S rO )r �'SECURITY_UBUNTU_STANDARD_UPDATES_POCKET�STANDARD_UPDATES_POCKET�SECURITY_UA_INFRA_POCKET�ESM_INFRA_POCKET�SECURITY_UA_APPS_POCKET�ESM_APPS_POCKET)r[ r5 r5 r6 �get_pocket_short_name� s r. �esm_cache_updatedc C sL | r$|s$t |�js$t�� }|d u rdS t�tj�}|| }|jdkr$dS dS )NT� F) r � is_attachedr �get_apt_cache_datetimer �nowr �utc�days)r� r/ r �last_apt_updater3 �time_since_updater5 r5 r6 �_should_update_esm_cache� s ��� r8 )r� r r c C s t |�}tt�}d} t| |tt|�� ��d�} |r | �|� |dkr7tjj j | _ | j t jdtjj id� | jS tjj j | _ t||�} t| �� �D ]'\} } | dkrg| jtjdd� | D �| d �d � tjj j | _ qHt| ||�\}}qH|su| jS tjtjtj fD �]}|| }|| }d d� |D �}t |�}|s�|r�tj j j | _ | j t jtj j ||d �d� q}|tjk}t || |�r�z t!�"|� d } W n, t#y� } z tj$j%t&|dt'|��d�}| jtj(|j |j)d�d � W Y d }~nd }~ww t*||�\}}|�rtjj j | _ |D ]}| jtj+|j,|j-|j.||d�d � q�|tjk�rt|tjk�r"d}nd}t/|�j0�s7| j t j1d||d�d� nt/|�j2}|t3j4j k�rP| j t j1t5j6j |d�d� t7|�j8�pWg } | �rbdd� | D �ng } || v�rt| j t j9||d�d� | j t j:|||d�d� q}| jS )NFr� r rv r �releasedc S � g | ]\}}|�qS r5 r5 ��.0r �_r5 r5 r6 � <listcomp>4 s �z&_generate_fix_plan.<locals>.<listcomp>r� )r� rM c S r: r5 r5 r; r5 r5 r6 r> Q s )rv rZ r[ Tr� )r� r� )r) r� r( r� r[ r% r& r? rn )rl rZ c S s g | ]}|j �qS r5 )r# )r< rr r5 r5 r6 r> � s rs r] );r� r r� r� r� r r� r r� rd r� r� r� r+ r3 r8 r; r� r� r r� r� rB rD r r r( r* r, r. r� r: r8 r �update_esm_caches� Exception�E_UPDATING_ESM_CACHEr" �getattrrV rE r# r� rC r) r* r( r r1 r1 �contract_statusr �ACTIVEr> r@ r �enabled_servicesr2 r4 ) r� r r r� r� r r� �countr� r/ r� �pkg_status_groups� status_valuer� r� r[ � pkg_src_groupr� � source_pkgs� pocket_namer� r� r� r� r� � unfixed_pkg�service_to_check�contract_expiry_statusrE �enabled_services_namesr5 r5 r6 r s � � ���� ��� ����� ����� ���� �� ���� ���� r r� )a�enumr � collectionsr r r �typingr r r r r r �uaclientr r r �'uaclient.api.u.pro.security.fix._commonr r r r r r r r r r r r r r �-uaclient.api.u.pro.status.enabled_services.v1r �(uaclient.api.u.pro.status.is_attached.v1r r �uaclient.configr �uaclient.data_typesr r! r" r# r$ r) r+ r- rV r' �unique�Enumr+ r8 r> rB rG rX r` rk rp rq rt ru rz r{ r~ r� r� r� r� r� r� r� r� r� r� r� r� r� r� r� r� r� r� r� �boolr� r r r r r& r' r. r8 r r5 r5 r5 r6 �<module> s @ �� 3 Ks��� ��� ��� ��� � �� � � �1I ��� � ��� ���� � plan/__init__.py 0000644 00000000000 15030137052 0007571 0 ustar 00 execute/v1.py 0000644 00000023233 15030137052 0007105 0 ustar 00 from typing import List, Optional from uaclient import apt, messages, util from uaclient.api.u.pro.security.fix._common import FixStatus, status_message from uaclient.api.u.pro.security.fix._common.plan.v1 import ( FixPlanAptUpgradeStep, FixPlanAttachStep, FixPlanEnableStep, FixPlanNoOpStatus, FixPlanNoOpStep, FixPlanResult, FixPlanWarningPackageCannotBeInstalled, FixPlanWarningSecurityIssueNotFixed, ) from uaclient.data_types import DataObject, Field, StringDataValue, data_list class UpgradedPackage(DataObject): fields = [ Field("name", StringDataValue, doc="The name of the package"), Field( "version", StringDataValue, doc="The version that the package was upgraded to", ), Field( "pocket", StringDataValue, doc="The pocket which contained the package upgrade", ), ] def __init__(self, name: str, version: str, pocket: str): self.name = name self.version = version self.pocket = pocket class FailedUpgrade(DataObject): fields = [ Field("name", StringDataValue, doc="The name of the package"), Field( "pocket", StringDataValue, required=False, doc="The pocket which contained the package upgrade", ), ] def __init__(self, name: str, pocket: Optional[str] = None): self.name = name self.pocket = pocket class FixExecuteError(DataObject): fields = [ Field("error_type", StringDataValue, doc="The type of the error"), Field( "reason", StringDataValue, doc="The reason why the error occurred" ), Field( "failed_upgrades", data_list(FailedUpgrade), required=False, doc="A list of ``FailedUpgrade`` objects", ), ] def __init__( self, error_type: str, reason: str, failed_upgrades: Optional[List[FailedUpgrade]] = None, ): self.error_type = error_type self.reason = reason self.failed_upgrades = failed_upgrades class FixExecuteResult(DataObject): fields = [ Field("title", StringDataValue, doc="The title of the CVE"), Field( "description", StringDataValue, required=False, doc="The description of the CVE", ), Field("status", StringDataValue, doc="The status of fixing the CVE"), Field( "upgraded_packages", data_list(UpgradedPackage), required=False, doc="A list of ``UpgradedPackage`` objects", ), Field( "errors", data_list(FixExecuteError), required=False, doc="A list of ``FixExecuteError`` objects", ), ] def __init__( self, title: str, status: str, description: Optional[str] = None, upgraded_packages: Optional[List[UpgradedPackage]] = None, errors: Optional[List[FixExecuteError]] = None, ): self.title = title self.description = description self.status = status self.upgraded_packages = upgraded_packages self.errors = errors class ExecuteContext: def __init__(self): self.require_enable = False self.require_attach = False self.status = FixStatus.SYSTEM_NON_VULNERABLE.value.msg self.upgraded_pkgs = [] # type: List[UpgradedPackage] self.errors = [] # type: List[FixExecuteError] def _handle_error( execute_context: ExecuteContext, security_issue: FixPlanResult ): if security_issue.error: execute_context.errors.append( FixExecuteError( error_type=security_issue.error.code or "unexpected-error", reason=security_issue.error.msg, ) ) execute_context.status = "error" def _handle_security_issue_not_fixed( execute_context: ExecuteContext, warning: FixPlanWarningSecurityIssueNotFixed, ): execute_context.errors.append( FixExecuteError( error_type=warning.warning_type, reason=status_message(warning.data.status), failed_upgrades=[ FailedUpgrade(name=pkg) for pkg in warning.data.source_packages ], ) ) execute_context.status = FixStatus.SYSTEM_STILL_VULNERABLE.value.msg def _handle_package_cannot_be_installed( execute_context: ExecuteContext, warning: FixPlanWarningPackageCannotBeInstalled, ): execute_context.errors.append( FixExecuteError( error_type=warning.warning_type, reason=messages.FIX_CANNOT_INSTALL_PACKAGE.format( package=warning.data.binary_package, version=warning.data.binary_package_version, ), failed_upgrades=[ FailedUpgrade( name=warning.data.binary_package, pocket=warning.data.pocket, ) ], ) ) execute_context.status = FixStatus.SYSTEM_STILL_VULNERABLE.value.msg def _handle_attach(execute_context: ExecuteContext, step: FixPlanAttachStep): execute_context.errors.append( FixExecuteError( error_type="fix-requires-attach", reason=messages.SECURITY_UPDATE_NOT_INSTALLED_SUBSCRIPTION, failed_upgrades=[ FailedUpgrade(name=pkg, pocket=step.data.required_service) for pkg in step.data.source_packages ], ) ) execute_context.require_attach = True execute_context.status = FixStatus.SYSTEM_STILL_VULNERABLE.value.msg def _handle_enable(execute_context: ExecuteContext, step: FixPlanEnableStep): if execute_context.require_attach: return execute_context.errors.append( FixExecuteError( error_type="fix-requires-enable", reason=messages.SECURITY_SERVICE_DISABLED.format( service=step.data.service ), failed_upgrades=[ FailedUpgrade(name=pkg, pocket=step.data.service) for pkg in step.data.source_packages ], ) ) execute_context.require_enable = True execute_context.status = FixStatus.SYSTEM_STILL_VULNERABLE.value.msg def _handle_apt_upgrade( execute_context: ExecuteContext, step: FixPlanAptUpgradeStep ): if execute_context.require_attach or execute_context.require_enable: return if not step.data.binary_packages: return if not util.we_are_currently_root(): execute_context.errors.append( FixExecuteError( error_type="fix-require-root", reason=messages.SECURITY_APT_NON_ROOT, failed_upgrades=[ FailedUpgrade(name=pkg, pocket=step.data.pocket) for pkg in step.data.source_packages ], ) ) execute_context.status = "error" 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"}, ) for pkg in step.data.binary_packages: pkg_version = apt.get_pkg_version(pkg) if pkg_version: execute_context.upgraded_pkgs.append( UpgradedPackage( name=pkg, version=pkg_version, pocket=step.data.pocket, ) ) except Exception as e: msg = getattr(e, "msg", str(e)) execute_context.status = FixStatus.SYSTEM_STILL_VULNERABLE.value.msg execute_context.errors.append( FixExecuteError( error_type="fix-error-installing-pkg", reason=msg, failed_upgrades=[ FailedUpgrade(name=pkg, pocket=step.data.pocket) for pkg in step.data.source_packages ], ) ) def _handle_noop(execute_context: ExecuteContext, step: FixPlanNoOpStep): if step.data.status == FixPlanNoOpStatus.NOT_AFFECTED.value: execute_context.status = FixStatus.SYSTEM_NOT_AFFECTED.value.msg def _execute_fix(security_issue: FixPlanResult) -> FixExecuteResult: execute_context = ExecuteContext() if security_issue.error: _handle_error(execute_context, security_issue) if security_issue.warnings: for warning in security_issue.warnings: if isinstance(warning, FixPlanWarningSecurityIssueNotFixed): _handle_security_issue_not_fixed(execute_context, warning) elif isinstance(warning, FixPlanWarningPackageCannotBeInstalled): _handle_package_cannot_be_installed(execute_context, warning) if security_issue.plan: for step in security_issue.plan: if isinstance(step, FixPlanAttachStep): _handle_attach(execute_context, step) elif isinstance(step, FixPlanEnableStep): _handle_enable(execute_context, step) elif isinstance(step, FixPlanAptUpgradeStep): _handle_apt_upgrade(execute_context, step) elif isinstance(step, FixPlanNoOpStep): _handle_noop(execute_context, step) return FixExecuteResult( title=security_issue.title, description=security_issue.description, status=execute_context.status, upgraded_packages=execute_context.upgraded_pkgs, errors=None if not execute_context.errors else execute_context.errors, ) execute/__pycache__/__init__.cpython-310.pyc 0000644 00000000267 15030137052 0014657 0 ustar 00 o �ϴf � @ s d S )N� r r r �Z/usr/lib/python3/dist-packages/uaclient/api/u/pro/security/fix/_common/execute/__init__.py�<module> s execute/__pycache__/v1.cpython-310.pyc 0000644 00000017224 15030137052 0013447 0 ustar 00 o �(�f�& � @ sV d dl mZmZ d dlmZmZmZ d dlmZm Z d dl m Z m Z m Z mZmZmZmZmZ d dlmZmZmZmZ G dd� de�ZG dd � d e�ZG d d � d e�ZG d d � d e�ZG dd� d�Z de defdd�Z de defdd�Z de defdd�Z de de fdd�Z de de fd d �Z!de de fd d �Z"de defd d!�Z#ded"efd#d$�Z$d%S )&� )�List�Optional)�apt�messages�util)� FixStatus�status_message)�FixPlanAptUpgradeStep�FixPlanAttachStep�FixPlanEnableStep�FixPlanNoOpStatus�FixPlanNoOpStep� FixPlanResult�&FixPlanWarningPackageCannotBeInstalled�#FixPlanWarningSecurityIssueNotFixed)� DataObject�Field�StringDataValue� data_listc @ sJ e Zd Zededd�ededd�ededd�gZdededefdd �Zd S ) �UpgradedPackage�name�The name of the package��doc�versionz,The version that the package was upgraded to�pocket�.The pocket which contained the package upgradec C � || _ || _|| _d S �N�r r r )�selfr r r � r! �T/usr/lib/python3/dist-packages/uaclient/api/u/pro/security/fix/_common/execute/v1.py�__init__! s zUpgradedPackage.__init__N)�__name__� __module__� __qualname__r r �fields�strr# r! r! r! r" r s ���r c @ sB e Zd Zededd�ededdd�gZd dedee fd d �ZdS ) � FailedUpgrader r r r Fr ��requiredr Nc C s || _ || _d S r �r r )r r r r! r! r" r# 2 s zFailedUpgrade.__init__r ) r$ r% r& r r r' r( r r# r! r! r! r" r) ' s �� r) c @ s\ e Zd Zededd�ededd�edee�ddd �gZ d dedede e e fd d �Z d S )�FixExecuteError� error_typezThe type of the errorr �reasonz!The reason why the error occurred�failed_upgradesFz#A list of ``FailedUpgrade`` objectsr* Nc C r r �r. r/ r0 )r r. r/ r0 r! r! r" r# E s zFixExecuteError.__init__r ) r$ r% r& r r r r) r' r( r r r# r! r! r! r" r- 7 s( ������ �r- c @ s� e Zd Zededd�ededdd�eded d�ed ee�dd d�ed ee�dd d�gZ dde de de e d e e e d e e e f dd�Z dS )�FixExecuteResult�titlezThe title of the CVEr � descriptionFzThe description of the CVEr* �statusz The status of fixing the CVE�upgraded_packagesz%A list of ``UpgradedPackage`` objects�errorsz%A list of ``FixExecuteError`` objectsNc C s" || _ || _|| _|| _|| _d S r �r3 r4 r5 r6 r7 )r r3 r5 r4 r6 r7 r! r! r" r# h s zFixExecuteResult.__init__)NNN) r$ r% r& r r r r r- r' r( r r r# r! r! r! r" r2 P sH � ������� � �r2 c @ s e Zd Zdd� ZdS )�ExecuteContextc C s( d| _ d| _tjjj| _g | _g | _d S )NF) �require_enable�require_attachr �SYSTEM_NON_VULNERABLE�value�msgr5 � upgraded_pkgsr7 )r r! r! r" r# x s zExecuteContext.__init__N)r$ r% r& r# r! r! r! r" r9 w s r9 �execute_context�security_issuec C s4 |j r| j�t|j jp d|j jd�� d| _d S d S )Nzunexpected-error)r. r/ �error)rB r7 �appendr- �coder> r5 )r@ rA r! r! r" � _handle_error� s �� �rE �warningc C s> | j �t|jt|jj�dd� |jjD �d�� tj j j | _d S )Nc S s g | ]}t |d ��qS ))r )r) ��.0�pkgr! r! r" � <listcomp>� s �z4_handle_security_issue_not_fixed.<locals>.<listcomp>r1 ) r7 rC r- � warning_typer �datar5 �source_packagesr �SYSTEM_STILL_VULNERABLEr= r> �r@ rF r! r! r" � _handle_security_issue_not_fixed� s ��� rP c C sN | j �t|jtjj|jj|jj d�t |jj|jj d�gd�� t j jj| _d S )N)�packager r, r1 )r7 rC r- rK r �FIX_CANNOT_INSTALL_PACKAGE�formatrL �binary_package�binary_package_versionr) r r rN r= r> r5 rO r! r! r" �#_handle_package_cannot_be_installed� s �����rV �stepc s@ | j �tdtj� fdd�� jjD �d�� d| _tj j j | _ d S )Nzfix-requires-attachc � g | ] }t |� jjd ��qS �r, )r) rL �required_servicerG �rW r! r" rJ � � ��z"_handle_attach.<locals>.<listcomp>r1 T) r7 rC r- r �*SECURITY_UPDATE_NOT_INSTALLED_SUBSCRIPTIONrL rM r; r rN r= r> r5 �r@ rW r! r[ r" �_handle_attach� s ��� r_ c sV | j rd S | j�tdtjj� jjd�� fdd�� jj D �d�� d| _ t j j j| _d S )Nzfix-requires-enable)�servicec rX rY )r) rL r` rG r[ r! r" rJ � r\ z"_handle_enable.<locals>.<listcomp>r1 T)r; r7 rC r- r �SECURITY_SERVICE_DISABLEDrS rL r` rM r: r rN r= r> r5 r^ r! r[ r" �_handle_enable� s � ��� rb c s& | j s| jrd S � jjsd S t�� s,| j�tdt j � fdd�� jj D �d�� d| _ d S z1t �� t jg d�� jj ddid � � jjD ]}t �|�}|rZ| j�t||� jjd �� qDW d S ty� } z)t|d t|��}tjjj| _ | j�td |� fd d�� jj D �d�� W Y d }~d S d }~ww )Nzfix-require-rootc rX rY �r) rL r rG r[ r! r" rJ � r\ z'_handle_apt_upgrade.<locals>.<listcomp>r1 rB )zapt-get�installz--only-upgradez-y�DEBIAN_FRONTEND�noninteractive)�cmd�override_env_varsr r> zfix-error-installing-pkgc rX rY rc rG r[ r! r" rJ r\ )r; r: rL �binary_packagesr �we_are_currently_rootr7 rC r- r �SECURITY_APT_NON_ROOTrM r5 r �run_apt_update_command�run_apt_command�get_pkg_versionr? r r � Exception�getattrr( r rN r= r> )r@ rW rI � pkg_version�er> r! r[ r" �_handle_apt_upgrade� sd ��� �� ���� �����rs c C s$ |j jtjjkrtjjj| _d S d S r )rL r5 r � NOT_AFFECTEDr= r �SYSTEM_NOT_AFFECTEDr> r^ r! r! r" � _handle_noop s �rv �returnc C s� t � }| jr t|| � | jr)| jD ]}t|t�r t||� qt|t�r(t||� q| j r]| j D ]-}t|t �r<t ||� q/t|t �rGt ||� q/t|t�rRt||� q/t|t�r\t||� q/t| j| j|j|j|jsmd d�S |jd�S )Nr8 )r9 rB rE �warnings� isinstancer rP r rV �planr r_ r rb r rs r rv r2 r3 r4 r5 r? r7 )rA r@ rF rW r! r! r" � _execute_fix s< � ���r{ N)%�typingr r �uaclientr r r �'uaclient.api.u.pro.security.fix._commonr r �/uaclient.api.u.pro.security.fix._common.plan.v1r r r r r r r r �uaclient.data_typesr r r r r r) r- r2 r9 rE rP rV r_ rb rs rv r{ r! r! r! r" �<module> sD ( ' � � � �� �� �: execute/__init__.py 0000644 00000000000 15030137052 0010301 0 ustar 00 __init__.py 0000644 00000064315 15030137052 0006662 0 ustar 00 import copy import enum import socket from collections import defaultdict from typing import Any, Dict, List, NamedTuple, Optional, Tuple from uaclient import apt, exceptions, livepatch, messages, system, util from uaclient.http import serviceclient CVE_OR_USN_REGEX = ( r"((CVE|cve)-\d{4}-\d{4,7}$|(USN|usn|LSN|lsn)-\d{1,5}-\d{1,2}$)" ) API_V1_CVES = "cves.json" API_V1_CVE_TMPL = "cves/{cve}.json" API_V1_NOTICES = "notices.json" API_V1_NOTICE_TMPL = "notices/{notice}.json" STANDARD_UPDATES_POCKET = "standard-updates" ESM_INFRA_POCKET = "esm-infra" ESM_APPS_POCKET = "esm-apps" BinaryPackageFix = NamedTuple( "BinaryPackageFix", [ ("source_pkg", str), ("binary_pkg", str), ("fixed_version", str), ], ) UnfixedPackage = NamedTuple( "UnfixedPackage", [ ("pkg", str), ("unfixed_reason", str), ], ) class FixStatus(enum.Enum): """ An enum to represent the system status after fix operation """ class _Value: def __init__(self, value: int, msg: str): self.value = value self.msg = msg SYSTEM_NON_VULNERABLE = _Value(0, "fixed") SYSTEM_NOT_AFFECTED = _Value(0, "not-affected") SYSTEM_STILL_VULNERABLE = _Value(1, "still-affected") SYSTEM_VULNERABLE_UNTIL_REBOOT = _Value(2, "affected-until-reboot") @property def exit_code(self): return self.value.value def __str__(self): return self.value.msg class UASecurityClient(serviceclient.UAServiceClient): url_timeout = 20 cfg_url_base_attr = "security_url" def _get_query_params( self, query_params: Dict[str, Any] ) -> Dict[str, Any]: """ Update query params with data from feature config. """ extra_security_params = self.cfg.cfg.get("features", {}).get( "extra_security_params", {} ) if query_params: query_params.update(extra_security_params) return query_params return extra_security_params @util.retry(socket.timeout, retry_sleeps=[1, 3, 5]) def request_url( self, path, data=None, headers=None, method=None, query_params=None ): query_params = self._get_query_params(query_params) return super().request_url( path=path, data=data, headers=headers, method=method, query_params=query_params, log_response_body=False, ) def get_cves( self, query: Optional[str] = None, priority: Optional[str] = None, package: Optional[str] = None, limit: Optional[int] = None, offset: Optional[int] = None, component: Optional[str] = None, version: Optional[str] = None, status: Optional[List[str]] = None, ) -> List["CVE"]: """Query to match multiple-CVEs. @return: List of CVE instances based on the the JSON response. """ query_params = { "q": query, "priority": priority, "package": package, "limit": limit, "offset": offset, "component": component, "version": version, "status": status, } response = self.request_url(API_V1_CVES, query_params=query_params) if response.code != 200: raise exceptions.SecurityAPIError( url=API_V1_CVES, code=response.code, body=response.body ) return [ CVE(client=self, response=cve_md) for cve_md in response.json_list ] def get_cve(self, cve_id: str) -> "CVE": """Query to match single-CVE. @return: CVE instance for JSON response from the Security API. """ url = API_V1_CVE_TMPL.format(cve=cve_id) response = self.request_url(url) if response.code != 200: raise exceptions.SecurityAPIError( url=url, code=response.code, body=response.body ) return CVE(client=self, response=response.json_dict) def get_notices( self, cves: Optional[str] = None, release: Optional[str] = None, limit: Optional[int] = None, offset: Optional[int] = None, order: Optional[str] = None, ) -> List["USN"]: """Query to match multiple-USNs. @return: Sorted list of USN instances based on the the JSON response. """ query_params = { "cves": cves, "release": release, "limit": limit, "offset": offset, "order": order, } response = self.request_url(API_V1_NOTICES, query_params=query_params) if response.code != 200: raise exceptions.SecurityAPIError( url=API_V1_NOTICES, code=response.code, body=response.body ) return sorted( [ USN(client=self, response=usn_md) for usn_md in response.json_dict.get("notices", []) if (cves is None or cves in usn_md.get("cves_ids", [])) and usn_md.get("id", "").startswith("USN-") ], key=lambda x: x.id, ) def get_notice(self, notice_id: str) -> "USN": """Query to match single-USN. @return: USN instance representing the JSON response. """ url = API_V1_NOTICE_TMPL.format(notice=notice_id) response = self.request_url(url) if response.code != 200: raise exceptions.SecurityAPIError( url=url, code=response.code, body=response.body ) return USN(client=self, response=response.json_dict) # Model for Security API responses class CVEPackageStatus: """Class representing specific CVE PackageStatus on an Ubuntu series""" def __init__(self, cve_response: Dict[str, Any]): self.response = cve_response @property def description(self): return self.response["description"] @property def fixed_version(self): return self.description @property def pocket(self): return self.response["pocket"] @property def release_codename(self): return self.response["release_codename"] @property def status(self): return self.response["status"] @property def status_message(self): if self.status == "needed": return messages.SECURITY_CVE_STATUS_NEEDED elif self.status == "needs-triage": return messages.SECURITY_CVE_STATUS_TRIAGE elif self.status == "pending": return messages.SECURITY_CVE_STATUS_PENDING elif self.status in ("ignored", "deferred"): return messages.SECURITY_CVE_STATUS_IGNORED elif self.status == "DNE": return messages.SECURITY_CVE_STATUS_DNE elif self.status == "not-affected": return messages.SECURITY_CVE_STATUS_NOT_AFFECTED elif self.status == "released": return messages.SECURITY_FIX_RELEASE_STREAM.format( fix_stream=self.pocket_source ) return messages.SECURITY_CVE_STATUS_UNKNOWN.format(status=self.status) @property def requires_ua(self) -> bool: """Return True if the package requires an active Pro subscription.""" return bool( self.pocket_source != messages.SECURITY_UBUNTU_STANDARD_UPDATES_POCKET ) @property def pocket_source(self): """Human-readable string representing where the fix is published.""" if self.pocket == "esm-infra": fix_source = messages.SECURITY_UA_INFRA_POCKET elif self.pocket == "esm-apps": fix_source = messages.SECURITY_UA_APPS_POCKET elif self.pocket in ("updates", "security"): fix_source = messages.SECURITY_UBUNTU_STANDARD_UPDATES_POCKET else: # TODO(GH: #1376 drop this when esm* pockets supplied by API) if "esm" in self.fixed_version: fix_source = messages.SECURITY_UA_INFRA_POCKET else: fix_source = messages.SECURITY_UBUNTU_STANDARD_UPDATES_POCKET return fix_source class CVE: """Class representing CVE response from the SecurityClient""" def __init__(self, client: UASecurityClient, response: Dict[str, Any]): self.response = response self.client = client def __eq__(self, other) -> bool: if not isinstance(other, CVE): return False return self.response == other.response @property def id(self): return self.response.get("id", "UNKNOWN_CVE_ID").upper() @property def notices_ids(self) -> List[str]: return self.response.get("notices_ids", []) @property def notices(self) -> List["USN"]: """Return a list of USN instances from API response 'notices'. Cache the value to avoid extra work on multiple calls. """ if not hasattr(self, "_notices"): self._notices = sorted( [ USN(self.client, notice) for notice in self.response.get("notices", []) if notice and notice.get("id", "").startswith("USN-") ], key=lambda n: n.id, reverse=True, ) return self._notices @property def description(self): return self.response.get("description") @property def packages_status(self) -> Dict[str, CVEPackageStatus]: """Dict of package status dicts for the current Ubuntu series. Top-level keys are source packages names and each value is a CVEPackageStatus object """ if hasattr(self, "_packages_status"): return self._packages_status # type: ignore self._packages_status = {} series = system.get_release_info().series for package in self.response["packages"]: for pkg_status in package["statuses"]: if pkg_status["release_codename"] == series: self._packages_status[package["name"]] = CVEPackageStatus( pkg_status ) return self._packages_status class USN: """Class representing USN response from the SecurityClient""" def __init__(self, client: UASecurityClient, response: Dict[str, Any]): self.response = response self.client = client def __eq__(self, other) -> bool: if not isinstance(other, USN): return False return self.response == other.response @property def id(self) -> str: return self.response.get("id", "UNKNOWN_USN_ID").upper() @property def cves_ids(self) -> List[str]: """List of CVE IDs related to this USN.""" return self.response.get("cves_ids", []) @property def cves(self) -> List[CVE]: """List of CVE instances based on API response 'cves' key. Cache the values to avoid extra work for multiple call-sites. """ if not hasattr(self, "_cves"): self._cves = sorted( [ CVE(self.client, cve) for cve in self.response.get("cves", []) ], key=lambda n: n.id, reverse=True, ) # type: List[CVE] return self._cves @property def title(self): return self.response.get("title") @property def references(self): return self.response.get("references") @property def release_packages(self) -> Dict[str, Dict[str, Dict[str, str]]]: """Binary package information available for this release. Reformat the USN.release_packages response to key it based on source package name and related binary package names. :return: Dict keyed by source package name. The second-level key will be binary package names generated from that source package and the values will be the dict response from USN.release_packages for that binary package. The binary metadata contains the following keys: name, version. Optional additional keys: pocket and component. """ if hasattr(self, "_release_packages"): return self._release_packages series = system.get_release_info().series self._release_packages = {} # type: Dict[str, Dict[str, Any]] # Organize source and binary packages under a common source package key for pkg in self.response.get("release_packages", {}).get(series, []): if pkg.get("is_source"): # Create a "source" key under src_pkg_name with API response if pkg["name"] in self._release_packages: if "source" in self._release_packages[pkg["name"]]: raise exceptions.SecurityAPIMetadataError( error_msg=( "{usn} metadata defines duplicate source" " packages {pkg}" ).format(usn=self.id, pkg=pkg["name"]), issue=self.id, extra_info="", ) self._release_packages[pkg["name"]]["source"] = pkg else: self._release_packages[pkg["name"]] = {"source": pkg} else: # is_source == False or None, then this is a binary package. # If processed before a source item, the top-level key will # not exist yet. # TODO(GH: 1465: determine if this is expected on kern pkgs) if not pkg.get("source_link"): raise exceptions.SecurityAPIMetadataError( error_msg=( "{issue} metadata does not define release_packages" " source_link for {bin_pkg}." ).format(issue=self.id, bin_pkg=pkg["name"]), issue=self.id, extra_info="", ) elif "/" not in pkg["source_link"]: raise exceptions.SecurityAPIMetadataError( error_msg=( "{issue} metadata has unexpected release_packages" " source_link value for {bin_pkg}: {link}" ).format( issue=self.id, bin_pkg=pkg["name"], link=pkg["source_link"], ), issue=self.id, extra_info="", ) source_pkg_name = pkg["source_link"].split("/")[-1] if source_pkg_name not in self._release_packages: self._release_packages[source_pkg_name] = {} self._release_packages[source_pkg_name][pkg["name"]] = pkg return self._release_packages def get_cve_affected_source_packages_status( cve: CVE, installed_packages: Dict[str, Dict[str, str]] ) -> Dict[str, CVEPackageStatus]: """Get a dict of any CVEPackageStatuses affecting this Ubuntu release. :return: Dict of active CVEPackageStatus keyed by source package names. """ affected_pkg_versions = {} for source_pkg, package_status in cve.packages_status.items(): if package_status.status == "not-affected": continue if source_pkg in installed_packages: affected_pkg_versions[source_pkg] = package_status return affected_pkg_versions def query_installed_source_pkg_versions() -> Dict[str, Dict[str, str]]: """Return a dict of all source packages installed on the system. The dict keys will be source package name: "krb5". The value will be a dict with keys binary_pkg and version. """ status_field = "${db:Status-Status}" out, _err = system.subp( [ "dpkg-query", "-f=${Package},${Source},${Version}," + status_field + "\n", "-W", ] ) installed_packages = {} # type: Dict[str, Dict[str, str]] for pkg_line in out.splitlines(): pkg_name, source_pkg_name, pkg_version, status = pkg_line.split(",") if not source_pkg_name: # some package don't define the Source source_pkg_name = pkg_name if "installed" not in status: continue if source_pkg_name in installed_packages: installed_packages[source_pkg_name][pkg_name] = pkg_version else: installed_packages[source_pkg_name] = {pkg_name: pkg_version} return installed_packages def get_related_usns(usn, client): """For a give usn, get the related USNs for it. For each CVE associated with the given USN, we capture other USNs that are related to the CVE. We consider those USNs related to the original USN. """ # If the usn does not have any associated cves on it, # we cannot establish a relation between USNs if not usn.cves: return [] related_usns = {} for cve in usn.cves: for related_usn_id in cve.notices_ids: # We should ignore any other item that is not a USN # For example, LSNs if not related_usn_id.startswith("USN-"): continue if related_usn_id == usn.id: continue if related_usn_id not in related_usns: related_usns[related_usn_id] = client.get_notice( notice_id=related_usn_id ) return list(sorted(related_usns.values(), key=lambda x: x.id)) def get_affected_packages_from_cves(cves, installed_packages): affected_pkgs = {} # type: Dict[str, CVEPackageStatus] for cve in cves: for pkg_name, pkg_status in get_cve_affected_source_packages_status( cve, installed_packages ).items(): if pkg_name not in affected_pkgs: affected_pkgs[pkg_name] = pkg_status else: current_ver = affected_pkgs[pkg_name].fixed_version if ( apt.version_compare(current_ver, pkg_status.fixed_version) > 0 ): affected_pkgs[pkg_name] = pkg_status return affected_pkgs def get_affected_packages_from_usn(usn, installed_packages): affected_pkgs = {} # type: Dict[str, CVEPackageStatus] for pkg_name, pkg_info in usn.release_packages.items(): if pkg_name not in installed_packages: continue cve_response = defaultdict(str) cve_response["status"] = "released" # Here we are assuming that the pocket will be the same one across # all the different binary packages. all_pockets = { pkg_bin_info["pocket"] for _, pkg_bin_info in pkg_info.items() if pkg_bin_info.get("pocket") } if not all_pockets: raise exceptions.SecurityAPIMetadataError( error_msg=( "{} metadata defines no pocket information for " "any release packages." ).format(usn.id), issue=usn.id, extra_info="", ) cve_response["pocket"] = all_pockets.pop() affected_pkgs[pkg_name] = CVEPackageStatus(cve_response=cve_response) return affected_pkgs def get_usn_affected_packages_status( usn: USN, installed_packages: Dict[str, Dict[str, str]] ) -> Dict[str, CVEPackageStatus]: """Walk CVEs related to a USN and return a dict of all affected packages. :return: Dict keyed on source package name, with active CVEPackageStatus for the current Ubuntu release. """ if usn.cves: return get_affected_packages_from_cves(usn.cves, installed_packages) else: return get_affected_packages_from_usn(usn, installed_packages) def override_usn_release_package_status( pkg_status: CVEPackageStatus, usn_src_released_pkgs: Dict[str, Dict[str, str]], ) -> CVEPackageStatus: """Parse release status based on both pkg_status and USN.release_packages. Since some source packages in universe are not represented in CVEPackageStatus, rely on presence of such source packages in usn_src_released_pkgs to represent package as a "released" status. :param pkg_status: the CVEPackageStatus for this source package. :param usn_src_released_pkgs: The USN.release_packages representing only this source package. Normally, release_packages would have data on multiple source packages. :return: Tuple of: human-readable status message, boolean whether released, boolean whether the fix requires access to UA """ usn_pkg_status = copy.deepcopy(pkg_status) if usn_src_released_pkgs and usn_src_released_pkgs.get("source"): usn_pkg_status.response["status"] = "released" usn_pkg_status.response["description"] = usn_src_released_pkgs[ "source" ]["version"] for pkg_name, usn_released_pkg in usn_src_released_pkgs.items(): # Copy the pocket from any valid binary package pocket = usn_released_pkg.get("pocket") if pocket: usn_pkg_status.response["pocket"] = pocket break return usn_pkg_status def group_by_usn_package_status(affected_pkg_status, usn_released_pkgs): status_groups = {} # type: Dict[str, List[Tuple[str, CVEPackageStatus]]] for src_pkg, pkg_status in sorted(affected_pkg_status.items()): usn_released_src = usn_released_pkgs.get(src_pkg, {}) usn_pkg_status = override_usn_release_package_status( pkg_status, usn_released_src ) status_group = usn_pkg_status.status.replace("ignored", "deferred") if status_group not in status_groups: status_groups[status_group] = [] status_groups[status_group].append((src_pkg, usn_pkg_status)) return status_groups def merge_usn_released_binary_package_versions( usns: List[USN], beta_pockets: Dict[str, bool] ) -> Dict[str, Dict[str, Dict[str, str]]]: """Walk related USNs, merging the released binary package versions. For each USN, iterate over release_packages to collect released binary package names and required fix version. If multiple related USNs require different version fixes to the same binary package, track the maximum version required across all USNs. :param usns: List of USN response instances from which to calculate merge. :param beta_pockets: Dict keyed on service name: esm-infra, esm-apps the values of which will be true of USN response instances from which to calculate merge. :return: Dict keyed by source package name. Under each source package will be a dict with binary package name as keys and binary package metadata as the value. """ usn_pkg_versions = {} for usn in usns: # Aggregate USN.release_package binary versions into usn_pkg_versions for src_pkg, binary_pkg_versions in usn.release_packages.items(): public_bin_pkg_versions = { bin_pkg_name: bin_pkg_md for bin_pkg_name, bin_pkg_md in binary_pkg_versions.items() if False is beta_pockets.get(bin_pkg_md.get("pocket", "None"), False) } if src_pkg not in usn_pkg_versions and public_bin_pkg_versions: usn_pkg_versions[src_pkg] = public_bin_pkg_versions elif src_pkg in usn_pkg_versions: # Since src_pkg exists, only record this USN's binary version # when it is greater than the previous version in usn_src_pkg. usn_src_pkg = usn_pkg_versions[src_pkg] for bin_pkg, binary_pkg_md in public_bin_pkg_versions.items(): if bin_pkg not in usn_src_pkg: usn_src_pkg[bin_pkg] = binary_pkg_md else: prev_version = usn_src_pkg[bin_pkg]["version"] current_version = binary_pkg_md["version"] if ( apt.version_compare(current_version, prev_version) > 0 ): # binary_version is greater than prev_version usn_src_pkg[bin_pkg] = binary_pkg_md return usn_pkg_versions def _check_cve_fixed_by_livepatch( issue_id: str, ) -> Tuple[Optional[FixStatus], Optional[str]]: # Check livepatch status for CVE in fixes before checking CVE api lp_status = livepatch.status() if ( lp_status is not None and lp_status.livepatch is not None and lp_status.livepatch.fixes is not None ): for fix in lp_status.livepatch.fixes: if fix.name == issue_id.lower() and fix.patched: version = lp_status.livepatch.version or "N/A" return (FixStatus.SYSTEM_NON_VULNERABLE, version) return (None, None) def status_message(status, pocket_source: Optional[str] = None): if status == "needed": return messages.SECURITY_CVE_STATUS_NEEDED elif status == "needs-triage": return messages.SECURITY_CVE_STATUS_TRIAGE elif status == "pending": return messages.SECURITY_CVE_STATUS_PENDING elif status in ("ignored", "deferred"): return messages.SECURITY_CVE_STATUS_IGNORED elif status == "DNE": return messages.SECURITY_CVE_STATUS_DNE elif status == "not-affected": return messages.SECURITY_CVE_STATUS_NOT_AFFECTED elif status == "released" and pocket_source: return messages.SECURITY_FIX_RELEASE_STREAM.format( fix_stream=pocket_source ) return messages.SECURITY_CVE_STATUS_UNKNOWN.format(status=status) def get_expected_overall_status( current_fix_status: str, fix_status: str ) -> str: if not current_fix_status: return fix_status if fix_status in ( FixStatus.SYSTEM_NON_VULNERABLE.value.msg, FixStatus.SYSTEM_NOT_AFFECTED.value.msg, ): if ( current_fix_status == FixStatus.SYSTEM_NOT_AFFECTED.value.msg and current_fix_status != fix_status ): return fix_status else: return current_fix_status else: # This means the system is still affected and we must # priotize this as the global state return fix_status
| ver. 1.4 |
Github
|
.
| PHP 8.2.28 | Generation time: 0.02 |
proxy
|
phpinfo
|
Settings