File manager - Edit - /home/newsbmcs.com/public_html/static/img/logo/Core.tar
Back
__pycache__/__init__.cpython-310.pyc 0000644 00000000232 15030661120 0013203 0 ustar 00 o ��{f � @ s d S )N� r r r �=/usr/lib/python3/dist-packages/UpdateManager/Core/__init__.py�<module> s __pycache__/roam.cpython-310.pyc 0000644 00000011044 15030661120 0012405 0 ustar 00 o ��{f� � @ s� d dl mZ d dlZd dlZG dd� de�ZG dd� de�Zedkr[ejdd� rAejd d krAe� Z e d e � � � e d e � � � e� Z e �� Ze d e� erTe�d� e�d � dS dS ) � )�print_functionNc @ sl e Zd ZdZed ZdZdZdZdZdZ dZ dZ dZ dZ dZdZdZd d � Zed d � �Zd d� Zdd� ZdS )�ModemManagerHelper� org.freedesktop.ModemManagerz.Modem� � r � � � c C s8 t �� | _| j�dd�| _t �| j| j�}|�� | _d S )Nr z /org/freedesktop/ModemManager) �dbus� SystemBus�bus� get_object�proxy� Interface� MM_DBUS_IFACE�EnumerateDevices�modems)�self� modem_manager� r �9/usr/lib/python3/dist-packages/UpdateManager/Core/roam.py�__init__C s �zModemManagerHelper.__init__c C � t �| d�}|�||�}|S �Nz org.freedesktop.DBus.Properties�r r �Get�r � interface�property�propsr r r �get_dbus_propertyJ � z$ModemManagerHelper.get_dbus_propertyc C sp | j D ]2}| j�| j|�}| �|| jd�}|| jkrqt�|| jd �}|� � }|d | j | j fv r5 dS qdS )N�Typez .Gsm.Networkr TF) r r r r r �MM_DBUS_IFACE_MODEM�MM_MODEM_TYPE_GSMr r �GetRegistrationInfo�'MM_MODEM_GSM_NETWORK_REG_STATUS_UNKNOWN�'MM_MODEM_GSM_NETWORK_REG_STATUS_ROAMING)r �m�dev�type�net�regr r r �is_gsm_roamingP s" � � ��z!ModemManagerHelper.is_gsm_roamingc C s� | j D ];}| j�| j|�}| �|| jd�}|| jkrqt�|| jd �}|� � \}}| j | j f}||v r7 dS ||v r> dS qdS )Nr" z.CdmaTF) r r r r r r# �MM_MODEM_TYPE_CDMAr r �GetRegistrationState�+MM_MODEM_CDMA_REGISTRATION_STATE_REGISTERED�(MM_MODEM_CDMA_REGISTRATION_STATE_ROAMING)r r( r) r* �cdma�cmda_1x�evdo�roaming_statesr r r �is_cdma_roamingb s$ � ��z"ModemManagerHelper.is_cdma_roamingN)�__name__� __module__� __qualname__r r# r$ r. �$MM_MODEM_GSM_NETWORK_REG_STATUS_IDLE�$MM_MODEM_GSM_NETWORK_REG_STATUS_HOME�)MM_MODEM_GSM_NETWORK_REG_STATUS_SEARCHING�&MM_MODEM_GSM_NETWORK_REG_STATUS_DENIEDr& r' �(MM_MODEM_CDMA_REGISTRATION_STATE_UNKNOWNr0 �%MM_MODEM_CDMA_REGISTRATION_STATE_HOMEr1 r � staticmethodr r- r6 r r r r r s( r c @ sx e Zd ZdZdZdZdZeegZdZdZ dZ dZ ee e e gZ dZ d ZdZdZd Zd d � Zed d� �Zdd� Zdd� ZdS )�NetworkManagerHelper� org.freedesktop.NetworkManagerr r �( r �2 �< �F r r c C s t �� | _| j�dd�| _d S )NrB z /org/freedesktop/NetworkManager)r r r r r )r r r r r � s �z NetworkManagerHelper.__init__c C r r r r r r r r � r! z&NetworkManagerHelper.get_dbus_propertyc C s� d}| � | j| jd�}|D ]J}| j�| j|�}| � || jd d�}|s$q | � || jd d�}|D ]&}| j�| j|�}| � || jd d�} | | jkrM dS | | jkrV dS q0q |S ) NF�ActiveConnectionsz.Connection.Active�Default�Devicesz.Device� DeviceTypeT)r r � NM_DBUS_IFACEr r �NM_DEVICE_TYPE_GSM�NM_DEVICE_TYPE_CDMA) r �res�actives�a�active� default_route�devs�dr) r* r r r � is_active_connection_gsm_or_cdma� s4 � � � � � z5NetworkManagerHelper.is_active_connection_gsm_or_cdmac C s. d}| � � rt� }||�� O }||�� O }|S )NF)rU r r- r6 )r rN �mmhelperr r r �(is_active_connection_gsm_or_cdma_roaming� s z=NetworkManagerHelper.is_active_connection_gsm_or_cdma_roamingN)r7 r8 r9 rK �NM_STATE_UNKNOWN�NM_STATE_CONNECTING_OLD�NM_STATE_CONNECTING�NM_STATE_CONNECTING_LIST�NM_STATE_CONNECTED_OLD�NM_STATE_CONNECTED_LOCAL�NM_STATE_CONNECTED_SITE�NM_STATE_CONNECTED_GLOBAL�NM_STATE_CONNECTED_LIST�NM_DEVICE_TYPE_UNKNOWN�NM_DEVICE_TYPE_ETHERNET�NM_DEVICE_TYPE_WIFIrL rM r r@ r rU rW r r r r rA x s6 �� rA �__main__r z--testr- �is_cdma_romaingzroam: )� __future__r r �sys�objectr rA r7 �argvrV �printr- r6 �nmhelperrW � is_roaming�exitr r r r �<module> s" YN � __pycache__/MetaRelease.cpython-310.pyc 0000644 00000022407 15030661120 0013643 0 ustar 00 o ��{f�D � @ s~ d dl mZmZ d dlZd dlZd dlZzd dlZW n ey' d dlZY nw zd dl m Z W n ey? d dl m Z Y nw d dl Z d dl Zd dlZd dlZd dlZd dlZd dlZzd dlmZ d dlmZmZ d dlmZmZ W n ey� d dl mZmZmZmZmZ Y nw dd l m Z m Z m Z m!Z!m"Z" G d d � d e#�Z$G d d � d e%�Z&G dd� de%�Z'e(dkr�e'dd�Z)dS dS )� )�absolute_import�print_functionN)� BadStatusLine)�quote)�Request�urlopen)� HTTPError�URLError)r r r r r � )�get_lang�get_dist�get_dist_version�get_ubuntu_flavor�get_ubuntu_flavor_namec @ s e Zd ZdS )�MetaReleaseParseErrorN)�__name__� __module__� __qualname__� r r �@/usr/lib/python3/dist-packages/UpdateManager/Core/MetaRelease.pyr 6 s r c @ s e Zd Zdd� ZdS )�Distc C s: || _ || _|| _|| _d | _d | _d | _d | _d | _d S �N) �name�version�date� supported�releaseNotesURI�releaseNotesHtmlUri� upgradeTool�upgradeToolSig�upgrade_broken)�selfr r r r r r r �__init__; s z Dist.__init__N)r r r r" r r r r r : s r c @ s| e Zd ZdZdejv ZdZdZ ddd�Z d d � Z d d � Z d d� Z dd� Z dd� Zedd� �Zdd� Zdd� ZdS )�MetaReleaseCorezT A MetaReleaseCore object abstracts the list of released distributions. �DEBUG_UPDATE_MANAGERz$/etc/update-manager/release-upgradesz /etc/update-manager/meta-releaseFNc C s� |rd| _ | �d||f � || _|| _t�� | _d | _d | _|d u r't � � }t |d�| _ t |d�| _t� | _t� | _d | _d | _d}|d | _|d | _d| _d| _t�� }tj � | j �r�z|� | j � W n tj y� } zt!j"�#d | j | f � W Y d } ~ d S d } ~ ww |�$d �r�d } d D ]$} |�%| | �r�| �d | j |�&| | �f � t'| d | | f |�&| | �� q�t�� }tj � | j(��r,z|� | j(� W n tj y� } zt!j"�#d | j(| f � W Y d } ~ d S d } ~ ww |�%dd��r,|�&dd��)� } | dk�s| dk�r d| _| j�*� d S | dk�r)d| _t+�,� } | �-| j��r#| j| _n | �d� nd| _|�r3| j| _|�r?| j| j7 _n |�rJ| j| j7 _| �d| j � d | _.| �/� �sa| �d� d S tj0| j1d��2� d S )NTz1MetaRelease.__init__() useDevel=%s useProposed=%s)�cachez https://changelogs.ubuntu.com/z meta-releasezmeta-release-ltsz -developmentz -proposedz ERROR: failed to read '%s': %s� METARELEASE)�URI�URI_LTS�URI_UNSTABLE_POSTFIX�URI_PROPOSED_POSTFIXz%s: %s z%s_%s�DEFAULT�Prompt�never�no�ltsz Prompt=lts for non-LTS, ignoring�normalzmetarelease-uri: %sz _buildMetaReleaseFile failed)�target)3�DEBUG�_debug� forceDownload�useDevelopmentRelease� threading�Event� downloaded� upgradable_to�new_dist�apt�Cacher �flavorr � flavor_namer �current_dist_namer �current_dist_version�no_longer_supported�prompt�METARELEASE_URI�METARELEASE_URI_LTS� METARELEASE_URI_UNSTABLE_POSTFIX� METARELEASE_URI_PROPOSED_POSTFIX� configparser� ConfigParser�os�path�exists�CONF_METARELEASE�read�Error�sys�stderr�write� has_section� has_option�get�setattr�CONF�lower�set� distro_info�UbuntuDistroInfo�is_lts�metarelease_information�_buildMetaReleaseFile�Thread�download�start)r! r5 � useProposed�debug�forceLTSr4 r% �base_uri�parser�e�sec�krB �dir r r r" T s� � ��� � ���� zMetaReleaseCore.__init__c C s� t j�dt j�| j��| _z t| jd��� W n� ty� t � dt j� d��}t j� |�d }t j� |�sZzt � |� W n tyY } ztj�d| � W Y d }~Y dS d }~ww t j� |�s�zt �|d� W n ty� } ztj�d| � W Y d }~Y dS d }~ww t j�|d �}t j� |�s�zt �|� W n ty� } ztj�d| � W Y d }~Y dS d }~ww t j�|t j�| j��| _Y nw zt j�| j�dkr�t �| j� W d S W d S ty� Y d S w ) Nz/var/lib/update-manager/�a�XDG_CACHE_HOMEz~/.cacher zmkdir() failed: '%s'Fi� zupdate-manager-coreT)rI rJ �join�basenamerC �METARELEASE_FILE�open�close�IOError�getenv� expanduser�splitrK �makedirs�OSErrorrO rP rQ �mkdir�getsize�unlink� Exception)r! � cache_dir�cache_parent_dirrf rJ r r r r] � sf � � �� �� �� �� � ��z%MetaReleaseCore._buildMetaReleaseFilec C � || _ dS )z\ virtual function that is called when the distro is no longer supported N)rA �r! �distr r r �dist_no_longer_supported� � z(MetaReleaseCore.dist_no_longer_supportedc C r} )z\ virtual function that is called when a new distro release is available N)r: r~ r r r �new_dist_available� r� z"MetaReleaseCore.new_dist_availablec C sV | � d� | j}| � d| � d }g }t�| j�}z�|�� r�dD ] }||jvr-td| ��q |jd }| � d| � |jd }tt j � |��}d|d < t � t|��} t|jd �} |jd } t|| | | �} d |jv r~|jd | _t� } | r~| jd | 7 _d|jv r�|jd | _| �| �}|r�| j|7 _d|jv r�|jd | _d|jv r�|jd | _d|jv r�|jd | _|�| � ||kr�| }|�� s W n tjy� td| j ��w | j�� d | _|d u r�| � d� dS d}|D ]} | j |j k�r | j s�| j s�q�| }| � d| � nq�|dk�r |j �s || _ | � |� |dk�r)|| _ | �!|� dS )NzMetaRelease.parse()zcurrent dist name: '%s')r �Version� Supported�DatezRequired key '%s' missingr zfound distro name: '%s'r� r � r� r� � ReleaseNotesz?lang=%s�ReleaseNotesHtml� UpgradeTool�UpgradeToolSignature� UpgradeBrokenzUnable to parse %sz,current dist not found in meta-release file F� z new dist: %sT)"r3 r? �apt_pkg�TagFiler\ �step�sectionr �list�email�utils� parsedate�time�mktime�tuple�intr r r r �#_get_release_notes_uri_query_stringr r r �appendrN rC rp r r r5 r9 r� r� )r! r? � current_dist�dists� index_tag� required_keyr �rawdate� parseddater r r r �lang�queryr9 r r r �parse� s� �� � � ��$�� � zMetaReleaseCore.parsec C s� | � d� d}t| j�}|�dd� |�dd� t�| jtj�r3z t�| j�j }W n t y2 Y nw |dkrF| j sF|�dt � t �|��� zmt|dd �}tj�| j�r�t�| jtj�s�zt�| j� W n t y } ztd | j|f � W Y d }~nd }~ww z$t| jd �}|�� D ] }|�|�d �� q�|�� |�dd� || _W n ty� Y nw |� � W nY t y� } z)|j d kr�tj�| j�r�| � d| j � t| jd�| _n| � d| � W Y d }~n)d }~w t t t!j"f�y } z| � d| � td| j � W Y d }~nd }~ww | jd u�rG| � d� z| �#� W n t$�y6 t%�&d| j � t�'| j� Y nw | j(�rF| j(j)�rFt�'| j� n| � d� | j*�+� d S )NzMetaRelease.download()r z Cache-ControlzNo-Cache�Pragmazno-cachezIf-Modified-Since� )�timeoutzCan't unlink '%s' (%s)zw+zUTF-8i0 zreading file '%s'�rz%result of meta-release download: '%s'zIFailed to connect to %s. Check your Internet connection or proxy settingsz!have self.metarelease_informationzparse failed for '%s'z NO self.metarelease_information),r3 r rC � add_headerrI �accessrn �W_OK�stat�st_mtimerv r4 r� �asctime�gmtimer rJ rK ry �printro � readlinesrQ �decode�flush�seekr\ rq rp r �coder r �socketr� r� rz �logging� exception�remover: r r8 rX )r! � lastmodified�req�urirf �f�liner r r r_ N s� �� ���� � ���� �� � zMetaReleaseCore.downloadc C s | j �� S r )r8 �is_set)r! r r r � downloading� s zMetaReleaseCore.downloadingc C s@ d}t � }|r |d| 7 }|d| j 7 }|d|j 7 }t|d�S )N�?zlang=%s&zos=%s&zver=%sz/?)r r= r r )r! r �qr� r r r r� � s z3MetaReleaseCore._get_release_notes_uri_query_stringc C s | j r tj�|d � d S d S )N� )r2 rO rP rQ )r! �msgr r r r3 � s �zMetaReleaseCore._debug)FFFFFN)r r r �__doc__rI �environr2 rV rL r" r] r� r� r� r_ �propertyr� r� r3 r r r r r# H s* �h.XG r# �__main__F)*� __future__r r r; r� rY rG � ImportErrorrH � http.clientr �httplibr� � email.utilsr� rI r� rO r� r6 � urllib.parser �urllib.requestr r � urllib.errorr r �urllib2r� r r r r r rz r �objectr r# r �metar r r r �<module> sJ � � � e� __pycache__/UpdateList.cpython-310.pyc 0000644 00000040160 15030661120 0013526 0 ustar 00 o ��{fV` � @ s� d dl mZ d dlZe�dde� d dlmZ d dlZd dlZd dl Z d dl Z d dl Z d dl Z d dl Z d dlZd dlmZ d dlmZ G dd � d �ZG d d � d e�ZG d d � d e�ZG dd� de�ZG dd� de�ZG dd� d�ZG dd� d�ZdS )� )�print_functionN�ignorez Accessed deprecated property)�gettext)�Gio)�utilsc @ s e Zd Zddd�Zdd� ZdS )� UpdateItemTc C s" || _ || _|| _|| _|| _d S �N)�icon�name�pkg� to_remove� sensitive��selfr r r r r � r �?/usr/lib/python3/dist-packages/UpdateManager/Core/UpdateList.py�__init__/ s zUpdateItem.__init__c C s | j s | jjp | jjS | jjS r )r r �marked_install�marked_upgrade� marked_delete�r r r r � is_selected6 s zUpdateItem.is_selectedN�T)�__name__� __module__� __qualname__r r r r r r r . s r c @ sn e Zd Zi Zddd�Zedd� �Z ddd �Zd d � Zd d � Z dd� Z ddd�Z dd� Z dd� Z dd� ZdS )� UpdateGroupTc C sZ t �| |||||� t� | _t� | _d | _|d ur+t |||||�| _| j�| j� d S d S r )r r �set�_items�_deps� core_item�addr r r r r @ s �zUpdateGroup.__init__c C s g }|� | j� t|dd� d�S )Nc S � | j �� S r �r �lower��ar r r �<lambda>M � z#UpdateGroup.items.<locals>.<lambda>��key)�extendr �sorted)r � all_itemsr r r �itemsI s zUpdateGroup.itemsNFc C sx t �|�}tj�d�}| j�t|||||�� |r6|j|v r8|j| j vr:| j s-| � ||� | � |||� d S d S d S d S �N�package) r �get_package_labelr � ThemedIcon�newr r! r r r � _init_deps� _add_deps)r r �cache�eventloop_callbackr r r r r r r r! O s �zUpdateGroup.addc C s || j v S r )r �r �itemr r r �contains^ s zUpdateGroup.containsc C s4 | j D ]}|jr|jj| jvr| �|j||� qd S r )r r r r r5 )r r6 r7 r9 r r r r4 a s ��zUpdateGroup._init_depsc C s( |du s|j du s|j| jv r&| s|j rdp d|j }t�d| � dS t| j�d dkr6t|�r6|� | j�|j� |j| jv r`| j|j D ]}|| jvr]||v r]| � || ||� qIdS |j }|� dd�}t j � |�D ]"}|j} | | jvr�| |v r�| j�|jg ��| � | � || ||� qodS ) z8Adds pkg and dependencies of pkg to the dependency list.Nz,Package was None or didn't have a candidate.z%s already in _deps.z Useless call to _add_deps. %s�� r �Depends� Recommends)� candidater r �logging�debug�len�callabler! � _depcacher5 �get_dependencies� itertools�chain� from_iterable� setdefault�append) r r r6 r7 �reason�depr> � dependencies�dependency_pkgr r r r r5 f s4 �� �� ��zUpdateGroup._add_depsc C s" | j s |r | �||� |j| j v S r )r r4 r )r � maybe_depr6 r7 r r r � is_dependency� s zUpdateGroup.is_dependencyc C s | j D ] }|�� r dS qdS )NTF)r. r r8 r r r �packages_are_selected� s �z!UpdateGroup.packages_are_selectedc C s. dd� | j D �}t|�dkot|�t| j �k S )Nc S s g | ]}|� � r|�qS r )r )�.0r9 r r r � <listcomp>� s z9UpdateGroup.selection_is_inconsistent.<locals>.<listcomp>r )r. rA )r �pkgs_installingr r r �selection_is_inconsistent� s �z%UpdateGroup.selection_is_inconsistentc C s2 | j rdS d}| jD ] }|t|jjdd�7 }q |S )Nr �size)r r. �getattrr r> )r rU r9 r r r �get_total_size� s zUpdateGroup.get_total_sizer )NNFT)NN)r r r rC r �propertyr. r! r: r4 r5 rO rP rT rW r r r r r = s � r c � e Zd Zd� fdd� Z� ZS )�UpdateApplicationGroupTc s, |� � }|�� }tt| ��|||||� d S r )�get_display_name�get_icon�superrZ r )r r � applicationr r r r �� __class__r r r � s �z UpdateApplicationGroup.__init__r �r r r r � __classcell__r r r_ r rZ � � rZ c rY )�UpdatePackageGroupTc s2 t �|�}tj�d�}tt| ��|||||� d S r/ )r r1 r r2 r3 r] rd r )r r r r r r r_ r r r � s �zUpdatePackageGroup.__init__r ra r r r_ r rd � rc rd c rY )�UpdateSystemGroupTc s< t d�tj|d� }tj�d�}tt| ��d ||||� d S )Nz%s base�r6 zdistributor-logo) �_r �get_ubuntu_flavor_namer r2 r3 r] re r )r r6 r r r r r_ r r r � s �zUpdateSystemGroup.__init__r ra r r r_ r re � rc re c @ � e Zd Zdd� ZdS )� UpdateOriginc C s g | _ || _|| _d S r )�packages� importance� description)r �descrl r r r r � s zUpdateOrigin.__init__N�r r r r r r r r rj � � rj c @ s� e Zd ZdZdZdZdZdZdZdZ d d d �Z d d � Z d d� Z dd� Z dd� Zdd� Zdd� Z d dd�Zdg g fdd �ZdS ) � UpdateListz} class that contains the list of available updates in self.pkgs[origin] where origin is the user readable string zPhased-Update-Percentagez/etc/machine-idz/var/lib/dbus/machine-idz+/usr/share/app-install/desktop/%s:*.desktopz-Update-Manager::Always-Include-Phased-Updatesz,Update-Manager::Never-Include-Phased-UpdatesNc C s< |r|nt �� | _d| _g | _g | _g | _g | _g | _g | _ g | _ d| _ t � � | _ g | _z t| j�� }|�� | _W d � n1 sBw Y W n# tyk t| j�� }|�� | _W d � n1 sdw Y Y nw dtjv r|tjd r|tjd }nd}dd� |�d�D �| _dtjv r�tj�d�| _nd| _i | _d S ) Nr � XDG_DATA_DIRSz /usr/local/share/:/usr/share/c S s g | ] }t j�|d ��qS )� applications)�os�path�join)rQ �baser r r rR � s �z'UpdateList.__init__.<locals>.<listcomp>�:�XDG_CURRENT_DESKTOP� ) r �get_dist�dist�distUpgradeWouldDelete� oem_groups� update_groups�security_groups�kernel_autoremove_groups�duplicate_groups�ubuntu_pro_groups�ubuntu_pro_fake_groups� num_updates�random�Random�ignored_phased_updates�open�UNIQ_MACHINE_ID_FILE�read�machine_uniq_id�FileNotFoundError� UNIQ_MACHINE_ID_FILE_FALLBACKrt �environ�split�application_dirs�get�current_desktop� desktop_cache)r �parentr| �f� data_dirsr r r r � sB �� ��� � zUpdateList.__init__c C s: |� d�sdS tj�|�}| jD ] }|�|�r dS qdS )Nz.desktopFT)�endswithrt ru �abspathr� � startswith)r � file_path�app_dirr r r �_file_is_application� s �z UpdateList._file_is_applicationc C sJ d}t j�|�� �}t j�|�d }|�� r#|d7 }||jkr#|d7 }|S )Nr � � )rt ru �basename� get_filename�splitext� should_showr )r r^ r �score� desktop_file�application_idr r r � _rate_application_for_package s z(UpdateList._rate_application_for_packagec C s� g }g }|j D ] }| �|�r|�|� q|j| jv r"|| j|j 7 }|D ]=}ztj�|�}|�| j � W n t yO } zt � d||f � W Y d }~q$d }~ww | � ||�}|dkra|�||f� q$|jdd� dd� t|�dkrw|d d S d S )Nz"Error loading .desktop file %s: %sr c S s | d S )Nr r )�appr r r r' , s z9UpdateList._get_application_for_package.<locals>.<lambda>T)r* �reverser� )�installed_filesr� rI r r� r �DesktopAppInfo�new_from_filename�set_desktop_envr� � Exceptionr? �warningr� �sortrA ) r r � desktop_files�rated_applications�installed_filer� r^ �er� r r r � _get_application_for_package s: � �� �� � z'UpdateList._get_application_for_packagec C s� |s t �d� d S t|�dkr| j|d }n| jd }t�|�D ]8}z|�d�d �d�d }W n tyA t �d| � Y q!w ||v rY| j � |g �� |� t � d ||f � q!d S ) Nz;_populate_desktop_cache called with empty list of packages.r� r �*�/���rx z7Could not extract package name from '%s'. File ignored.zApp candidate for %s: %s) r? r� rA �APP_INSTALL_PATTERN�glob�iglobr� � IndexError�errorr� rH rI r@ )r � pkg_names�patternr� r r r r �_populate_desktop_cache2 s, �����z"UpdateList._populate_desktop_cachec C s� | j sdS |jj}|jjD ]2}|r tj�|j|j�dkr q |jD ] \}}|j d| j kr>|j dkr>|j j � |�}|r> dS q q dS )a� This will test if the pkg is a security update. This includes if there is a newer version in -updates, but also an older update available in -security. For example, if installed pkg A v1.0 is available in both -updates (as v1.2) and -security (v1.1). we want to display it as a security update. :return: True if the update comes from the security pocket Fr z %s-security�UbuntuT)r| �_pkg� current_ver� version_list�apt�apt_pkg�version_compare�ver_str� file_list�archive�origin�_pcache�_list� find_index)r r �inst_ver�ver� verFileIter�index� indexfiler r r �_is_security_updateP s( �� ��z UpdateList._is_security_updatec C s� t jj�| jd�r dS | j|jjv rSt jj�| jd�r"t � d� dS | j � d|jj |jj| jf � |jj| j }| j �dd�}|t|�krSt � d|j||f � dS dS )z� This will test if the pkg is a phased update and if it needs to get installed or ignored. :return: True if the updates should be ignored Fz,holding back phased update per configurationTz%s-%s-%sr �d z'holding back phased update %s (%s < %s))r� r� �config�find_b� ALWAYS_INCLUDE_PHASED_UPDATES�PHASED_UPDATES_KEYr> �record� NEVER_INCLUDE_PHASED_UPDATESr? �infor� �seed� source_name�versionr� �randint�intr )r r � threshold� percentager r r �_is_ignored_phased_updatek s, �� � �z$UpdateList._is_ignored_phased_updateFTc C s� |sg S g }g }g }|D ] } | � | �} | d ur$t| | ||�} |�| � q |�| � q t|�D ]2} g } |D ]} | �| ||�rJ| �| � t| �dkrJ nq4t| �dkr`| d �| |||� |�| � q.d }|r�d}t� dd� dd� |D �� d �}t d d d ||�}t j |d �}|d d g}|D ]} |�| j�r�|�| j� q�|D ]} | |v r�|s�|�|| � q�|D ]"} |�| ||�r�|d u r�t|||�}|�| � q�|�t| ||�� q�|jd d � d� |jdd � d� |r�|�|� || S )Nr� r )zlinux$zlinux-.*-buildinfo.*zlinux-.*-dev.*zlinux-.*-generic.*zlinux-.*-headers.*zlinux-.*-hwe.*zlinux-.*-gcp.*zlinux-.*-kernel.*zlinux-.*-lowlatency.*zlinux-.*-modules.*zlinux-.*-raspi.*zlinux-.*-tools.*zlinux-.*-virtual.*z linux-base.*zlinux-crashdump.*z linux-doc.*�(�|c S s g | ]}d | �qS )�^r )rQ �nr r r rR � s z+UpdateList._make_groups.<locals>.<listcomp>�)rf zubuntu-standardzubuntu-minimalc S r" r r# r% r r r r' � r( z)UpdateList._make_groups.<locals>.<lambda>r) c S r" r r# r% r r r r' � r( )r� rZ rI �listrO rA r! �remove�re�compilerv r r �get_ubuntu_flavor_package�matchr re rd r� )r r6 �pkgsr7 r r �ungrouped_pkgs� app_groups� pkg_groupsr r� � app_group� dep_groups�group� system_group� linux_names� linux_regexp�ubuntu_base_group�flavor_package�ubuntu_base_pkgsr r r � _make_groups� s� � � � �� � � � ��� � zUpdateList._make_groupsc s� g | _ |�� | _g }g }g }g }g } g } G dd� d�� G � fdd�d�} g } g } |D ]\}}}}|r8| �|� q*| �| |||�� q*|D ]�}|jsL|jr�t|jdd �d u r\td|j � qD| � |�}|so| � |�ro| j �|� qDt � |j d�r{|�|� |j | v }|r�|�|� n |r�|�|� n|�|� | jd | _|jr�|js�|js�| j �|j � qD|jr�|jr�|j�|j �r�|j�|j �s�| �|� |j |v r�| �|� qD| D ]}|�� q�| D ]}|�� q�| j D ]}|�� q�|s�|s�|�rd d � || | | | D �}| �|� | �|||�| _| �|||�| _| �|||�| _| �|| |d �| _| �|| |d �| _ | j|| |d d �| _ | �|||�| _ d S )Nc @ ri )z8UpdateList.update.<locals>.FakeUbuntuProPackageCandidatec S s. || _ || _|| _|| _|| _d| _i | _d S �NF)r� �summaryrm r� rU � downloadabler� )r r� r� rU r r r r � s zAUpdateList.update.<locals>.FakeUbuntuProPackageCandidate.__init__Nro r r r r � FakeUbuntuProPackageCandidate� rp r c s( e Zd Z� fdd�Zdd� Zdd� ZdS )z/UpdateList.update.<locals>.FakeUbuntuProPackagec s0 || _ � |||�| _d| _d| _d| _g | _d S r� )r r> r r r r� )r � package_namer� rU �r r r r � s � z8UpdateList.update.<locals>.FakeUbuntuProPackage.__init__c S � d S r r r r r r � mark_install � z<UpdateList.update.<locals>.FakeUbuntuProPackage.mark_installc S r r r r r r r � mark_delete r z;UpdateList.update.<locals>.FakeUbuntuProPackage.mark_deleteN)r r r r r r r r r r �FakeUbuntuProPackage� s r �originsz1WARNING: upgradable but no candidate.origins?!?: z oem-*-metar� c S s g | ]}|j �qS r )r )rQ �pr r r rR J s z%UpdateList.update.<locals>.<listcomp>TF)r ) � held_back�saveDistUpgrader} rI � is_upgradabler rV r> �printr r� r� r� �fnmatchr� r �is_auto_removable� versioned_kernel_pkgs_regexpr� �running_kernel_pkgs_regexpr � mark_keepr� r� r~ r r� r� r� r� r� )r r6 r7 �duplicate_packages�ua_security_packages�oem_pkgs�pro_pkgs� security_pkgs� upgrade_pkgs�kernel_autoremove_pkgs�duplicate_pkgsr �fake_ua_packages�ua_packages_namesr r� rU r r �is_security_update�comes_from_pror� r r r �update� s� � � � �� ��� � ����� ������ �zUpdateList.updater )FT)r r r �__doc__r� r� r� r� r� r� r r� r� r� r� r� r� r� r r r r r rq � s. �� " �U�rq )� __future__r �warnings�filterwarnings�DeprecationWarningr rg r� r r? rE rt r� r� r� � gi.repositoryr �UpdateManager.Corer r r rZ rd re rj rq r r r r �<module> s. � a __pycache__/utils.cpython-310.pyc 0000644 00000035536 15030661120 0012623 0 ustar 00 o ��{fmH � @ sB d dl mZ d dlmZ d dlmZ d dlmZmZmZ d dl m Z d dl Z d dl Z e � � d dlZd dlZd dlZd dlZd dlZd dlZd dlZzd dlmZmZmZmZmZ d dlm Z W n e y} d dl mZmZmZmZmZ d dl m Z Y nw d d l m Z G d d � d e!�Z"G d d � d �Z#dd� Z$dd� Z%dMdd�Z&dd� Z'dNdd�Z(dd� Z)d d � Z*d d � Z+d d!� Z,d"d#� Z-d$d%� Z.G d&d'� d'e�Z/dMd(d)�Z0dMd*d+�Z1d,d-� Z2d.d/� Z3d0d1� Z4d2d3� Z5dMd4d5�Z6d6d7� Z7dMd8d9�Z8dMd:d;�Z9d<d=� Z:d>d?� Z;d@dA� Z<dBdC� Z=dDdE� Z>dFdG� Z?dHdI� Z@eAdJk�r e:ddKdL� dS dS )O� )�print_function)�gettext)�ngettext)�S_IMODE�ST_MODE�S_IXUSR)�ceilN)� ProxyHandler�Request� build_opener�install_opener�urlopen)�urlsplit)�copyc @ s* e Zd ZdZd dd�Zdd� Zdd� Zd S ) � ExecutionTimez� Helper that can be used in with statements to have a simple measure of the timing of a particular block of code, e.g. with ExecutionTime("db flush"): db.flush() � c C s || _ d S �N)�info)�selfr � r �:/usr/lib/python3/dist-packages/UpdateManager/Core/utils.py�__init__G s zExecutionTime.__init__c C s t � � | _d S r )�time�now�r r r r � __enter__J s zExecutionTime.__enter__c C s t d| jt�� | j f � d S )Nz%s: %s)�printr r r )r �type�value�stackr r r �__exit__M s zExecutionTime.__exit__N)r )�__name__� __module__� __qualname__�__doc__r r r r r r r r @ s r c @ s( e Zd ZdZdZdZdZdZdZdZ dS ) �SoftwarePropertiesPager � � � � � � N) r! r" r# �ubuntu_software�other_software�updates�authentication�additional_drivers�developer_options� ubuntu_pror r r r r% Q s r% c C sJ t | �}t|j�}|jr|j�|jd�|_|jr!|j�|jd�|_t|�S )Nzhidden-uzhidden-p)r r �uri�username�replace�password�str)�entry�tmp� url_partsr r r �)get_string_with_no_auth_from_source_entry[ s r; c C sD d} zddl }|�� }|�d�} W | S ty! t�d� Y | S w )z+ return True if Unity is currently running Fr Nzcom.canonical.Unityz&could not check for Unity dbus service)�dbus� SessionBus�name_has_owner� Exception�logging� exception)� unity_runningr< �busr r r �is_unity_runninge s � �rD c C s� |st �� }|dkrId| }t|�� }|�� }W d � n1 s!w Y |�d�d �d�d }|| kr8dS t|�d�d �� d �}|dks dS ) Nr z /proc/%s/stat�(r' �)Tr& F)�os�getpid�open�read� partition� rpartition�int�split)� processname�pid� stat_file�stat_f�stat�commandr r r �is_child_of_process_nameq s �� rU c C s0 t j�d�r t j�d�sdS t �d�t �d�kS )z, returns True if we are inside a chroot z/procz/proc/1T�/z /proc/1/root)rG �path�existsrS r r r r � inside_chroot� s rY �F r c C sL d}| � � D ] }t|�|�d� t|� |kr |d| 7 }||d 7 }q|S )z� helpers inspired after textwrap - unfortunately we can not use textwrap directly because it break packagenames with "-" in them into new lines r � � )rN �len�rfind)�t�width�subsequent_indent�out�sr r r �wrap� s rd c K s8 d}| � d�}|D ]}t|fi |��} || d 7 }q |S )Nr r[ )rN rd )rc �kwargs�msg�paras�parr r r �twrap� s ri c C sd g } t j�d�s g S tjdgtjdd�}|�� d �d�}|d= |D ] }|r/| �|�� d � q"| S )z= return list of loaded modules (or [] if lsmod is not found) z /sbin/lsmodT��stdout�universal_newlinesr r[ ) rG rW rX � subprocess�Popen�PIPE� communicaterN �append)�modules�p�lines�liner r r �lsmod� s ��rv c C sF t j�| �sdS tt �| �t �}|t@ tks!t �| |tB � dS dS )zB check if a given binary has the executable bit and if not, add itN)rG rW rX r rS r r �chmod)rW �moder r r �check_and_fix_xbit� s �ry c C s� ddi} dt jvr dS t jd �� }|dd� | v r"| |dd� S d|v rA|�d�d �d�d }d |v r=|�d �d }|d S |dd � d S ) z: helper to get the country mirror from the current locale �cr �LANGNr* �_�.r r& �@r' )rG �environ�lowerrN )� lang_mirror�lang�countryr r r �country_mirror� s r� c C � t j�d�} | rt�d| � | S ddlm}m} |g d�|dd�}|�� }|dkr4t j � d| � d S |j � � �� } |j �� | S ) z2 return the codename of the current runing distro �META_RELEASE_FAKE_CODENAMEzQusing fake release name '%s' (because of META_RELEASE_FAKE_CODENAME environment) r �rn ro )� lsb_releasez-c�-sTrj �"lsb_release returned exitcode: %i �unknown distribution�rG r �getr@ �warningrm rn ro �wait�sys�stderr�writerk �readline�strip�close)�distrn ro rs �resr r r �get_dist� �" � � r� c C r� ) z2 return the version of the current running distro �META_RELEASE_FAKE_VERSIONzSusing fake release version '%s' (because of META_RELEASE_FAKE_VERSION environment) r r� )r� z-rr� Trj r� r� r� )�descrn ro rs r� r r r �get_dist_version� r� r� c @ s e Zd Zdd� ZdS )� HeadRequestc C � dS )N�HEADr r r r r � get_method� s zHeadRequest.get_methodN)r! r" r# r� r r r r r� � s r� c C sT |sdd� |d| � t | �\}}}}}|d|||||f � |dv rSztt| ��}|�� |jdkr7W dS W dS tyR } z |d | � W Y d }~dS d }~ww |d kr�d d l} z1| �|�} | �� | � t j � |�� | � t j �|��} | �� |r�|d | � | d kr�W dS W dS ty� } z|r�|d| � W Y d }~dS d }~ww dS )z� helper that checks if the given uri exists and is downloadable (supports optional debug_func function handler to support e.g. logging) Supports http (via HEAD) and ftp (via size request) c S r� )NTr )�xr r r �<lambda> s z"url_downloadable.<locals>.<lambda>zurl_downloadable: %sz"s='%s' n='%s' p='%s' q='%s' f='%s')�http�https�� TFzerror from httplib: '%s'N�ftpr zftplib.size() returned: %szerror from ftplib: '%s')r r r� r� �coder? �ftplib�FTP�login�cwdrG rW �dirname�size�basename�quit) r3 � debug_func�scheme�netlocrW �querry�fragment� http_file�er� �fr� r r r �url_downloadable sP � �� �� ��r� c C sB d}i }t j�d�dkrt j�d�|d< n0tj�|�rEt �� }t �||� |�dd�}|rE|�d�}t |� d��}|rE|rEd ||f |d< t j�d �dkrVt j�d �|d < n d|v r`|d |d < |r�|� � D ]}t � d |�sztd | tjd� dS qft|�}t|�} t| � d|v r�t�d|d � d |v r�t�d|d � |S )z� init proxy settings * use apt.conf http proxy if present, * otherwise look into synaptics config file, * otherwise the default behavior will use http_proxy environment if present z /root/.synaptic/synaptic.confzAcquire::http::Proxyr r� zSynaptic::useProxyFzSynaptic::httpProxyzSynaptic::httpProxyPortz http://%s:%s/zAcquire::https::Proxyr� z https?://\w+zproxy '%s' looks invalid)�fileN� http_proxy� https_proxy)�apt_pkg�config�findrG rW rX � Configuration�read_config_file�find_br7 �find_i�values�re�matchr r� r� r r r �putenv) � gsettings�SYNAPTIC_CONF_FILE�proxies�cnf� use_proxy� proxy_host� proxy_port�proxy� proxy_support�openerr r r � init_proxy, s> �r� c C s� zOddl } | �| jj�}z|�dd�}| �|d�}|�dd�W W S | jjyO } z!d}|j|kr2� |�dd �}| �|d�}|�dd �W Y d}~W S d}~ww t yY Y d S w ) z� Check via dbus if the system is running on battery. This function is using UPower per default, if UPower is not available it falls-back to DeviceKit.Power. r Nzorg.freedesktop.UPowerz/org/freedesktop/UPowerz org.freedesktop.DBus.Properties� OnBatteryz)org.freedesktop.DBus.Error.ServiceUnknownz org.freedesktop.DeviceKit.Powerz /org/freedesktop/DeviceKit/Power� on_batteryF) r< �Bus� TYPE_SYSTEM� get_object� Interface�Get� exceptions� DBusException�_dbus_error_namer? )r< rC �devobj�devr� � error_unknownr r r r� X s. � � �� �r� c C sz z2ddl m} m} | �| jj�}|�dddd|�dd�d dd d d � \}}| j|� � |d d �}|W S t y< Y d S w ) z� Send a dbus signal to logind to not suspend the system, it will be released when the return value drops out of scope r )�Gio�GLibzorg.freedesktop.login1z/org/freedesktop/login1z org.freedesktop.login1.Manager�Inhibitz(ssss))zshutdown:sleep� UpdateManagerzUpdating System�blockN���)�fdF) � gi.repositoryr� r� � bus_get_sync�BusType�SYSTEM�call_with_unix_fd_list_sync�Variant�UnixInputStream� steal_fdsr? )r� r� � connection�var�fdlist� inhibitorr r r � inhibit_sleept s � � �r� c C s | dks | � � dkr dS dS )N�0�FALSEFT)�upper)r7 r r r � str_to_bool� s r� c C s: dd l } z t�� \}}|W S ty | �d� Y d S w )Nr zgedefaultlocale() failed)r@ �locale�getdefaultlocaler? rA )r@ �locale_s�encodingr r r �get_lang� s �r� c C s t | d�}|�dd�d S )z6 try to guess the flavor based on the running desktop ��cache�-r& r )�get_ubuntu_flavor_packagerN )r� �pkgr r r �get_ubuntu_flavor� s r c C s ddl m} | d�}|�dd�S )Nr )�DistUpgradeConfigz"/usr/share/ubuntu-release-upgrader�Distro�MetaPkgs)�#DistUpgrade.DistUpgradeConfigParserr �getlist)r �parserr r r �_load_meta_pkg_list� s r c C s� dg}z |� tt� �� W n ty$ } z td|� W Y d}~nd}~ww | du r-t�� } |D ]}|| v r9| | nd}|rD|jrD| S q/dS )zB try to guess the flavor metapackage based on the running desktop zubuntu-desktopz%Could not load list of meta packages:N)�extend�sortedr r? r �apt�Cache� is_installed)r� � meta_pkgsr� �meta_pkg� cache_pkgr r r r� � s �� �r� c C sb t | d�}ddi}||v r|| S |�d�r t|�dd�d �S |�d�r/t|�dd�d �S dS ) z; try to guess the flavor name based on the running desktop r� zubuntustudio-desktopz Ubuntu Studioz-desktopr& r z-netbook�Ubuntu)r� �endswith�capitalize_first_word�rsplit)r� r �lookupr r r �get_ubuntu_flavor_name� s r c C s� dd l }|�dd� ddlm}m} |j| |jj|jj |j j d�}|� d||f � |� � |�� �|jj� |�d� |�� |�� dS ) Nr �Gtkz3.0)r �Gdk)�parent�flagsr �buttonsz<big><b>%s</b></big> %sr F)�gi�require_versionr� r r � MessageDialog� DialogFlags�MODAL� MessageType�ERROR� ButtonsType�CLOSE� set_markup�realize� get_window� set_functions� WMFunction�MOVE� set_title�run�destroy)r �summary�messager r r �dr r r �error� s � r1 c C sH | dk rt t| td� ��}tdd|�d|i S t�td�| d d �S )zG Convert a given size in bytes to a nicer better readable unit i@B i� z %(size).0f kBr� z%.1f MBg @�@)rM r �floatr r� � format_stringr| )�bytes� size_in_kbr r r � humanize_size� s �r6 c C s t j�d�S )NzAPT::Architecture)r� r� r� r r r r �get_arch� s r7 c C s� d}d}d}t d��?}|�� D ]1}|�� }|sq|�� }|| }||kr%q|| �d�d }t|d�} | | kr@ W d � dS qW d � d S 1 sLw Y d S ) z@ check if the current system is listening on the given tcp port r& r( �0Az /proc/net/tcp�:� NTF)rI � readlinesr� rN rM ) �port�INDEX_LOCAL_ADDR� INDEX_STATE�STATE_LISTENING�net_tcpru r� �state�local_port_str� local_portr r r �is_port_already_listening� s. �� ��rD c C sV d} t �� dkr td��t j�d�sdS tjddgtjdd ��� d }|| kr)dS dS ) z# Return True if iptables is active z�Chain INPUT (policy ACCEPT) target prot opt source destination Chain FORWARD (policy ACCEPT) target prot opt source destination Chain OUTPUT (policy ACCEPT) target prot opt source destination r z%Need root to check the iptables statez/sbin/iptablesF�iptablesz-nLTrj ) rG �getuid�OSErrorrW rX rm rn ro rp )�iptables_emptyrb r r r �iptables_active s ��rI c C s@ t | �dkr | d �� r | d �� s | d �� | dd� S | S )z3 this uppercases the first word's first letter r& r N)r] �isalpha�isupper� capitalize)�stringr r r r , s $r c C s t | jdd�}t|�S )zX this takes a package synopsis and uppercases the first word's first letter r. r )�getattr� candidater )r �namer r r �get_package_label4 s rQ �__main__�bar�bazr )rZ r )B� __future__r r r| r rS r r r �mathr r r� � init_configr� r@ r� rG rm r� r �urllib.requestr r r r r � urllib.parser � ImportError�urllib2�urlparser �objectr r% r; rD rU rY rd ri rv ry r� r� r� r� r� r� r� r� r� r� r r r� r r1 r6 r7 rD rI r rQ r! r r r r �<module> sr � +, � __pycache__/MyCache.cpython-310.pyc 0000644 00000030127 15030661120 0012763 0 ustar 00 o ��{f�I � @ sZ d dl mZmZ d dlZe�dde� d dlZd dlZd dlZd dl Z zd dl m Z d dl m Z d dlmZ W n eyN d dlm Z m Z d dlmZ Y nw zd d lmZ W n eyf d d lmZ Y nw d dlZd dlZd dlZd dlZd d lmZ zd d l m Z W n ey� dZ Y nw d Z d Z e d Z!G dd� de"�Z#G dd� dej$j%�Z%dS )� )�absolute_import�print_functionN�ignorezapt API not stable yet)� HTTPError)�urlopen)�urlsplit)r r )� BadStatusLine)�gettext)� Launchpadz /var/lib/synaptic/preferencesz.https://changelogs.ubuntu.com/changelogs/pool/z%s/%s/%s/%s_%s/%sc @ s e Zd ZdZdS )� HttpsChangelogsUnsupportedErrorz� https changelogs with credentials are unsupported because of the lack of certitifcation validation in urllib2 which allows MITM attacks to steal the credentials N)�__name__� __module__� __qualname__�__doc__� r r �</usr/lib/python3/dist-packages/UpdateManager/Core/MyCache.pyr 9 s r c @ s� e Zd ZdZd.dd�Zdd� Zdd� Zd d � Zed d � �Z ed d� �Z dd� Z edd� �Z dd� Z dd� Zdd� Zdd� Z d/d d �Zd d � Zd d!� Zd"d#� Zd$d%� Zd&d'� Zd(d)� Zd*d+� Zd,d-� ZdS )0�MyCache�UbuntuNc s t j�| ||� || _t| j�dksJ �| �� rJ �| �� i | _i | _ i | _ | j j dkr1| � � | j j dkr=| j jdks?J �d | _tj�d�}|r}t�dd�dd� |D �� d �| _tjdd gd d ��� � t�dd�� fd d�|D �� d �| _d S d | _d | _d S ) Nr z APT::VersionedKernelPackages�(�|c S s g | ]}d | �qS )�^r ��.0�pr r r � <listcomp>` s z$MyCache.__init__.<locals>.<listcomp>�)�unamez-rT)�universal_newlinesc s. g | ]}|� d �sd| d � n� | �qS )z.*r )� startswithr ��running_kernel_versionr r r d s ���)�apt�Cache�__init__�rootdir�len�req_reinstall_pkgs�_dpkgJournalDirty� _initDepCache� all_changes�all_news� pro_versions� _depcache� broken_count�saveDistUpgrade� del_count� launchpad�apt_pkg�config� value_list�re�compile�join� versioned_kernel_pkgs_regexp� subprocess� check_output�rstrip�running_kernel_pkgs_regexp)�self�progressr$ �apt_versioned_kernel_pkgsr r r r# E sH � ���� �� � zMyCache.__init__c C s> t j�tj�d��d }t �|�D ] }t�d|�r dS qdS )z` test if the dpkg journal is dirty (similar to debSystem::CheckUpdates) zDir::State::statusz/updatesz[0-9]+TF) �os�path�dirnamer1 r2 � find_file�listdirr4 �match)r< �d�fr r r r' k s �� �zMyCache._dpkgJournalDirtyc C s0 | j �� tj�t�r| j �t� | j �� d S �N)r, � read_pinfiler? r@ �exists�SYNAPTIC_PINFILE�init�r< r r r r( w s zMyCache._initDepCachec C s | � � d S rG )r( rL r r r �clear� s z MyCache.clearc C s, t �| j�}t �� }|�|| j| j� |jS )z< get the size of the packages that are required to download )r1 �PackageManagerr, �Acquire� get_archives�_list�_records� fetch_needed)r< �pm�fetcherr r r �required_download� s zMyCache.required_downloadc C � | j jS rG )r, � inst_countrL r r r � install_count� � zMyCache.install_countc C rW rG )r, � keep_countrL r r r r[ � s zMyCache.keep_countc C rW rG )r, r/ rL r r r r/ � rZ zMyCache.del_countc C sR |D ]$}|sqd}|D ]}|j |jjks t�|j|j|j�s d}q |r& dS qdS )z<Return True if any of the dependencies in deps match target.TF)�name�package� shortnamer1 � check_dep�version�relation)r< �target�deps�dep_orrD �base_depr r r �_check_dependencies� s ����zMyCache._check_dependenciesc C sl |j }|sdS | D ]*}|j}|d ur3| �||�d��r3| �||�d��r3t�d|jj|jf � dS q dS )NF� Conflicts�Replacesz*%s Conflicts/Replaces %s; allowing removalT)� installed� candidaterf �get_dependencies�logging�infor] r^ )r< �pkgrb �cpkgrj r r r �find_removal_justification� s* � �� ���z"MyCache.find_removal_justificationc C s� | j �d� | j j}|dkr+dd� | D �}|t|�ks J �|D ] }| �|�r*|d8 }q |dkrB| �� | j jdkr?| j jdksAJ �n| j jdksJJ �| j �� |S )z@ this functions mimics a upgrade but will never remove anything Tr c S s g | ]}|j r|�qS r )� marked_delete)r rn r r r r � s z+MyCache.saveDistUpgrade.<locals>.<listcomp>� )r, �upgrader/ r% rp rM r- )r< � wouldDelete� deleted_pkgsrn r r r r. � s � zMyCache.saveDistUpgradec C s, |� d�}t|�dkrd�|dd� �}|S )z strip of the epoch �:rr � N)�splitr% r6 )r< �verstr� vers_no_epochr r r � _strip_epoch� s zMyCache._strip_epochFc C s� | | }|j j}d}|jj�|j�j}|j j} | �| �} |� d�} t | �dkr+| d }|d } |� d�r:d|d } |r?|} n t || ||| |f } t | �}|jdkrY|jrYtd��t| �}d }d t�|� } |�� �d d �}|d krv |S t�||�}|r�t|jdd�}|r�d|v r�|� dd�d }|�d�}|r�d|v r�|� dd�d }|r�|r�t�||�dk r� |S n |r�t�||�dkr� |S || }qg)z* helper that fetches the file in question �main�/rr r �lib� �httpszKhttps locations with username/password are notsupported to fetch changelogsrw z^%s \((.*)\)(.*)$TzUTF-8�replacer` Nrv ) rj � source_name�_pcacher, �get_candidate_ver�_pkg�section�source_versionr{ rx r% r �CHANGELOGS_URIr �scheme�usernamer r r4 �escape�readline�decoderD �getattrri �groupr1 �version_compare)r< r\ �fname�strict_versioning�changelogs_urirn �srcpkg� src_sectionr� � srcver_epoch�srcver� split_section�prefix�uri�res� changelog�alllines�regexp�linerD ri � changelogverr r r �_get_changelog_or_news� st �� � ��������z MyCache._get_changelog_or_newsc C s� t s t�d� dS | | j}|jD ]!}t|�jdkrqt�d|�}|dur2|� d�|� d�}} nqt� d� dS | j du rHt j dd d d �| _ | j j jd ||f d �}|du r_t� d� dS |j|jd|jd�}|srt� d� dS |d �� S )zbReturn the changelog URI from the Launchpad API Return None in case of an error. z9Launchpadlib not available, cannot retrieve PPA changelogNzppa.launchpad.netzhttp.*/(.*)/(.*)/ubuntu/.*rr � z)Unable to find a valid PPA candidate URL.zupdate-manager� production�devel)r` z ~%s/ubuntu/%s)� referencez6Unable to retrieve the archive from the Launchpad API.T)r� � exact_matchr` z;No published sources were retrieved from the Launchpad API.r )r rl �warningrj �urisr �hostnamer4 �searchr� �errorr0 �login_anonymously�archives�getByReference�getPublishedSourcesr� r� � changelogUrl) r< r\ �cdtr� rD �user�ppa�archive�spphsr r r �_extract_ppa_changelog_uri* sB � � � � z"MyCache._extract_ppa_changelog_uric C s� | | }|j j}|d u rd S |j j�d�}|sd S |�d�d �� }d|v r1|�d�d �d�}n|j j}|�d�d }|d||f S )N�Sourcer r rr r r} z/%s_%s.changelog) rj r� �record�getrx �stripr: r� � rpartition)r< r\ rn �deb_uri�srcrecr� r� �base_urir r r �+_guess_third_party_changelogs_uri_by_sourceX s z3MyCache._guess_third_party_changelogs_uri_by_sourcec C s, | | }|j j}|rd|�dd�d S dS )z` guess changelogs uri based on ArchiveURI by replacing .deb with .changelog z %s.changelog�.rr r N)rj r� �rsplit)r< r\ rn r� r r r �+_guess_third_party_changelogs_uri_by_binaryi s z3MyCache._guess_third_party_changelogs_uri_by_binaryc C s8 | � |� | �|� z|�� W d S ty Y d S w rG )�get_news� get_changelog�release� Exception)r< r\ �lockr r r �get_news_and_changelogu s �z MyCache.get_news_and_changelogc C s> z | � |dd�}W n ty Y dS w |r || j|< dS dS )z7 get the NEWS.Debian file from the changelogs location z NEWS.DebianTN)r� r� r* )r< r\ �newsr r r r� } s ��zMyCache.get_newsc C s� d }|D ] }|j �d�r#z| �|�}W n ty" t�d� Y qw q| �|�}| �|�}d}|||fD ]<}|rqz| �|dd|�} | j | | 7 < W q5 t t fy[ t d�}Y q5 t ttjfyp t�d� t d�}Y q5w q5| j | |7 < d S ) NzLP-PPA-z'Unable to connect to the Launchpad API.rw r� FzAThis update does not come from a source that supports changelogs.zerror on changelog fetching�OFailed to download the list of changes. Please check your Internet connection.)�originr r� r� rl � exceptionr� r� r� r) r r �_�IOErrorr �socketr� ) r< r\ �origins�changelogs_uri_ppar� �changelogs_uri_binary�changelogs_uri_source� error_messager� r� r r r �(_fetch_changelog_for_third_party_package� sJ �� ����� ���z0MyCache._fetch_changelog_for_third_party_packagec C s |D ] \}}}}|| j |< qd S rG )r+ )r< �pro_pkgsr\ r` �_ar r r �create_pro_cache� s �zMyCache.create_pro_cachec C s. | | j j}| j�|| | j j�}td�|t| | jdd�|f | j|< | j dd� |D �vr6| � ||� dS | | j j }| | j j � dd�}z| �|d�}t|�d kr[td �||f }W n/ tym td �||f }Y n tttjfy� } ztd |� td �}W Y d}~nd}~ww | j| |7 < dS )z4 get the changelog file from the changelog location zFChanges for %s versions: Installed version: %s Available version: %s r` Nc S s g | ]}|j �qS r )r� )r �or r r r � s z)MyCache.get_changelog.<locals>.<listcomp>rv z%3Ar� r z�The changelog does not contain any relevant changes. Please use http://launchpad.net/ubuntu/+source/%s/%s/+changelog until the changes become available or try again later.z�The list of changes is not available yet. Please use http://launchpad.net/ubuntu/+source/%s/%s/+changelog until the changes become available or try again later.zcaught exception: r� )rj r� r+ r� r` r� r� ri r) �CHANGELOG_ORIGINr� r� r� r� r� r% r r� r r� r� �print)r< r\ r� r` r� r� r� �er r r r� � s6 � �� � ��zMyCache.get_changelogrG )FN)r r r r� r# r' r( rM �propertyrV rY r[ r/ rf rp r. r{ r� r� r� r� r� r� r� r� r� r r r r r A s8 & �W. * r )&� __future__r r �warnings�filterwarnings� FutureWarningr! r1 rl r? � urllib.errorr �urllib.requestr � urllib.parser � ImportError�urllib2�urlparse� http.clientr �httplibr� r8 r4 � DistUpgrade.DistUpgradeCache� DistUpgrader r� �launchpadlib.launchpadr rJ �CHANGELOGS_POOLr� r� r �DistUpgradeCacher r r r r �<module> sF � � � __pycache__/AlertWatcher.cpython-310.pyc 0000644 00000005436 15030661120 0014044 0 ustar 00 o ��{f� � @ sB d dl mZ d dlmZ d dlZd dlmZ G dd� dej�ZdS )� )�absolute_import)�GObjectN)� DBusGMainLoopc @ s� e Zd ZdZejjdejffejjdejffejjdejejffd�Z dd� Z dd� Z dd � Z d d � Z d d � Zdd� Zdd� ZdS )� AlertWatcherzX a class that checks for alerts and reports them, like a battery or network warning N)� network-alert� battery-alert�network-3g-alertc C s0 t j �| � tdd� t�tjj�| _d| _d S )NT)�set_as_default� )r �__init__r �dbus�Bus� TYPE_SYSTEM�bus� network_state)�self� r �A/usr/lib/python3/dist-packages/UpdateManager/Core/AlertWatcher.pyr . s zAlertWatcher.__init__c C s� z>| j �dd�}|jd| jdd� t�|d�}|�dd�| _| �| j� | j �dd�}|jd | j dd� | � � | � � W d S tj j yJ Y d S w ) Nz org.freedesktop.NetworkManagerz /org/freedesktop/NetworkManager� StateChanged)�dbus_interface� org.freedesktop.DBus.Properties�State�org.freedesktop.UPower�/org/freedesktop/UPower�Changed) r � get_object�connect_to_signal�_on_network_state_changedr � Interface�Getr �_network_alert�_power_changed�_update_3g_state� exceptions� DBusException)r �obj� interfacer r r �check_alert_state5 s2 �� � � ��z AlertWatcher.check_alert_statec C s | � |� | �� d S )N)r r" �r �stater r r r L s z&AlertWatcher._on_network_state_changedc C s2 ddl m} |� }|�� }|�� }| �||� d S )N� )�NetworkManagerHelper)�roamr+ � is_active_connection_gsm_or_cdma�(is_active_connection_gsm_or_cdma_roaming�_network_3g_alert)r r+ �nm�on_3g� is_roamingr r r r" P s z AlertWatcher._update_3g_statec C s | � d||� d S )Nr )�emit)r r1 r2 r r r r/ W s z AlertWatcher._network_3g_alertc C s || _ | �d|� d S )Nr )r r3 r( r r r r Z s zAlertWatcher._network_alertc C s6 | j �dd�}t�|d�}|�dd�}| �d|� d S )Nr r r � OnBatteryr )r r r r r r3 )r r% r& � on_batteryr r r r! ^ s � zAlertWatcher._power_changed)�__name__� __module__� __qualname__�__doc__r � SignalFlags� RUN_FIRST�TYPE_INT� TYPE_BOOLEAN� __gsignals__r r' r r" r/ r r! r r r r r s0 ����� r )� __future__r � gi.repositoryr r �dbus.mainloop.glibr r r r r r �<module> s __pycache__/LivePatchSocket.cpython-310.pyc 0000644 00000010277 15030661120 0014506 0 ustar 00 o ��{f� � @ sh d dl mZ d dlZd dlZd dlZd dlZdZG dd� dejj �Z G dd� de �Z G dd � d e �Z dS ) � )�GLibNz5/var/snap/canonical-livepatch/current/livepatchd.sockc @ s e Zd Zdd� Zdd� ZdS )�UHTTPConnectionc C s t jj�| d� || _d S )N� localhost)�http�client�HTTPConnection�__init__�path)�selfr � r �D/usr/lib/python3/dist-packages/UpdateManager/Core/LivePatchSocket.pyr " s zUHTTPConnection.__init__c C s& t � t jt j�}|�| j� || _d S �N)�socket�AF_UNIX� SOCK_STREAM�connectr �sock)r r r r r r & s zUHTTPConnection.connectN)�__name__� __module__� __qualname__r r r r r r r s r c @ sB e Zd Zd dd�Zdd� Zedd� �Zedd � �Zed d � �ZdS ) �LivePatchSocketNc C s |d u r t t�| _d S || _d S r )r � HOST_NAME�conn)r � http_connr r r r . s zLivePatchSocket.__init__c s& � �fdd�}t j|d�}|�� d S )Nc s� z�j �dd� �j �� } | jdk� t�| �� �}W n ty( d� t� }Y nw t � |��t � |��t � |��t �� ����fdd�� d S )N�GETz/status?verbose=True�� Fc s �� ����S r r r )�active� check_state�fixes�on_done� patch_stater r �<lambda>B s �z=LivePatchSocket.get_status.<locals>.do_call.<locals>.<lambda>)r �request� getresponse�status�yaml� safe_load�read� Exception�dictr �get_check_state�get_patch_state� get_fixesr �idle_add)�r�data�r r )r r r r r �do_call6 s � z+LivePatchSocket.get_status.<locals>.do_call)�target)� threading�Thread�start)r r r1 �threadr r0 r � get_status4 s zLivePatchSocket.get_statusc C �@ z| d }t dd� |D �d �}|d d W S ty Y dS w )Nr$ c s � � | ] }|d r|V qdS ��runningNr ��.0�kr r r � <genexpr>L � � z2LivePatchSocket.get_check_state.<locals>.<genexpr>� livepatch� checkStatez check-failed��nextr( �r/ r$ �kernelr r r r* H � �z LivePatchSocket.get_check_statec C r8 )Nr$ c s r9 r: r r< r r r r? U r@ z2LivePatchSocket.get_patch_state.<locals>.<genexpr>rA � patchState�unknownrC rE r r r r+ Q rG z LivePatchSocket.get_patch_statec C s` z#| d }t dd� |D �d �}|d d }dd� |�dd ��d �D �W S ty/ t� Y S w ) Nr$ c s r9 r: r r< r r r r? ^ r@ z,LivePatchSocket.get_fixes.<locals>.<genexpr>rA r c S s g | ] }t |�d krt|��qS )r )�len� LivePatchFix)r= �fr r r � <listcomp>` s �z-LivePatchSocket.get_fixes.<locals>.<listcomp>z* � � )rD �replace�splitr( �list)r/ r$ rF r r r r r, Z s � �zLivePatchSocket.get_fixesr ) r r r r r7 � staticmethodr* r+ r, r r r r r , s r c @ s$ e Zd Zdd� Zdd� Zdd� ZdS )rK c C s& d}|� |�dk| _|�|d�| _d S )Nz (unpatched)���rN )�find�patchedrP �name)r �text�patched_patternr r r r h s zLivePatchFix.__init__c C s& t |t�r| j|jko| j|jkS tS r )� isinstancerK rW rV �NotImplemented)r �otherr r r �__eq__m s zLivePatchFix.__eq__c C s | � |�}|tu r |S | S r )r] r[ )r r\ �resultr r r �__ne__r s zLivePatchFix.__ne__N)r r r r r] r_ r r r r rK f s rK )� gi.repositoryr � http.clientr r r3 r% r r r r �objectr rK r r r r �<module> s : MetaRelease.py 0000644 00000042207 15030661120 0007304 0 ustar 00 # MetaRelease.py # -*- Mode: Python; indent-tabs-mode: nil; tab-width: 4; coding: utf-8 -*- # # Copyright (c) 2004,2005 Canonical # # Author: Michael Vogt <michael.vogt@ubuntu.com> # # This program is free software; you can redistribute it and/or # modify it under the terms of the GNU General Public License as # published by the Free Software Foundation; either version 2 of the # License, or (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program; if not, write to the Free Software # Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 # USA from __future__ import absolute_import, print_function import apt import apt_pkg import distro_info try: import configparser except ImportError: import ConfigParser as configparser try: from http.client import BadStatusLine except ImportError: from httplib import BadStatusLine import logging import email.utils import os import socket import sys import time import threading try: from urllib.parse import quote from urllib.request import Request, urlopen from urllib.error import HTTPError, URLError except ImportError: from urllib2 import HTTPError, Request, URLError, urlopen, quote from .utils import (get_lang, get_dist, get_dist_version, get_ubuntu_flavor, get_ubuntu_flavor_name) class MetaReleaseParseError(Exception): pass class Dist(object): def __init__(self, name, version, date, supported): self.name = name self.version = version self.date = date self.supported = supported self.releaseNotesURI = None self.releaseNotesHtmlUri = None self.upgradeTool = None self.upgradeToolSig = None # the server may report that the upgrade is broken currently self.upgrade_broken = None class MetaReleaseCore(object): """ A MetaReleaseCore object abstracts the list of released distributions. """ DEBUG = "DEBUG_UPDATE_MANAGER" in os.environ # some constants CONF = "/etc/update-manager/release-upgrades" CONF_METARELEASE = "/etc/update-manager/meta-release" def __init__(self, useDevelopmentRelease=False, useProposed=False, debug=False, forceLTS=False, forceDownload=False, cache=None): if debug: self.DEBUG = True self._debug("MetaRelease.__init__() useDevel=%s useProposed=%s" % (useDevelopmentRelease, useProposed)) # force download instead of sending if-modified-since self.forceDownload = forceDownload self.useDevelopmentRelease = useDevelopmentRelease # information about the available dists self.downloaded = threading.Event() self.upgradable_to = None self.new_dist = None if cache is None: cache = apt.Cache() self.flavor = get_ubuntu_flavor(cache=cache) self.flavor_name = get_ubuntu_flavor_name(cache=cache) self.current_dist_name = get_dist() self.current_dist_version = get_dist_version() self.no_longer_supported = None self.prompt = None # default (if the conf file is missing) base_uri = "https://changelogs.ubuntu.com/" self.METARELEASE_URI = base_uri + "meta-release" self.METARELEASE_URI_LTS = base_uri + "meta-release-lts" self.METARELEASE_URI_UNSTABLE_POSTFIX = "-development" self.METARELEASE_URI_PROPOSED_POSTFIX = "-proposed" # check the meta-release config first parser = configparser.ConfigParser() if os.path.exists(self.CONF_METARELEASE): try: parser.read(self.CONF_METARELEASE) except configparser.Error as e: sys.stderr.write("ERROR: failed to read '%s':\n%s" % ( self.CONF_METARELEASE, e)) return # make changing the metarelease file and the location # for the files easy if parser.has_section("METARELEASE"): sec = "METARELEASE" for k in ["URI", "URI_LTS", "URI_UNSTABLE_POSTFIX", "URI_PROPOSED_POSTFIX"]: if parser.has_option(sec, k): self._debug("%s: %s " % (self.CONF_METARELEASE, parser.get(sec, k))) setattr(self, "%s_%s" % (sec, k), parser.get(sec, k)) # check the config file first to figure if we want lts upgrades only parser = configparser.ConfigParser() if os.path.exists(self.CONF): try: parser.read(self.CONF) except configparser.Error as e: sys.stderr.write("ERROR: failed to read '%s':\n%s" % ( self.CONF, e)) return # now check which specific url to use if parser.has_option("DEFAULT", "Prompt"): prompt = parser.get("DEFAULT", "Prompt").lower() if (prompt == "never" or prompt == "no"): self.prompt = 'never' # nothing to do for this object # FIXME: what about no longer supported? self.downloaded.set() return elif prompt == "lts": self.prompt = 'lts' # the Prompt=lts setting only makes sense when running on # a LTS, otherwise it would result in users not receiving # any distro upgrades di = distro_info.UbuntuDistroInfo() if di.is_lts(self.current_dist_name): self.METARELEASE_URI = self.METARELEASE_URI_LTS else: self._debug("Prompt=lts for non-LTS, ignoring") else: self.prompt = 'normal' # needed for the _tryUpgradeSelf() code in DistUpgradeController if forceLTS: self.METARELEASE_URI = self.METARELEASE_URI_LTS # devel and proposed "just" change the postfix if useDevelopmentRelease: self.METARELEASE_URI += self.METARELEASE_URI_UNSTABLE_POSTFIX elif useProposed: self.METARELEASE_URI += self.METARELEASE_URI_PROPOSED_POSTFIX self._debug("metarelease-uri: %s" % self.METARELEASE_URI) self.metarelease_information = None if not self._buildMetaReleaseFile(): self._debug("_buildMetaReleaseFile failed") return # we start the download thread here and we have a timeout threading.Thread(target=self.download).start() #threading.Thread(target=self.check).start() def _buildMetaReleaseFile(self): # build the metarelease_file name self.METARELEASE_FILE = os.path.join( "/var/lib/update-manager/", os.path.basename(self.METARELEASE_URI)) # check if we can write to the global location, if not, # write to homedir try: open(self.METARELEASE_FILE, "a").close() except IOError: cache_dir = os.getenv( "XDG_CACHE_HOME", os.path.expanduser("~/.cache")) # Take special care when creating this directory; ~/.cache needs # to be created with mode 0700, but the other directories do # not. cache_parent_dir = os.path.split(cache_dir)[0] if not os.path.exists(cache_parent_dir): try: os.makedirs(cache_parent_dir) except OSError as e: sys.stderr.write("mkdir() failed: '%s'" % e) return False if not os.path.exists(cache_dir): try: os.mkdir(cache_dir, 0o700) except OSError as e: sys.stderr.write("mkdir() failed: '%s'" % e) return False path = os.path.join(cache_dir, 'update-manager-core') if not os.path.exists(path): try: os.mkdir(path) except OSError as e: sys.stderr.write("mkdir() failed: '%s'" % e) return False self.METARELEASE_FILE = os.path.join( path, os.path.basename(self.METARELEASE_URI)) # if it is empty, remove it to avoid I-M-S hits on empty file try: if os.path.getsize(self.METARELEASE_FILE) == 0: os.unlink(self.METARELEASE_FILE) except Exception: pass return True def dist_no_longer_supported(self, dist): """ virtual function that is called when the distro is no longer supported """ self.no_longer_supported = dist def new_dist_available(self, dist): """ virtual function that is called when a new distro release is available """ self.new_dist = dist def parse(self): self._debug("MetaRelease.parse()") current_dist_name = self.current_dist_name self._debug("current dist name: '%s'" % current_dist_name) current_dist = None dists = [] # parse the metarelease_information file index_tag = apt_pkg.TagFile(self.metarelease_information) try: while index_tag.step(): for required_key in ("Dist", "Version", "Supported", "Date"): if required_key not in index_tag.section: raise MetaReleaseParseError( "Required key '%s' missing" % required_key) name = index_tag.section["Dist"] self._debug("found distro name: '%s'" % name) rawdate = index_tag.section["Date"] parseddate = list(email.utils.parsedate(rawdate)) parseddate[8] = 0 # assume no DST date = time.mktime(tuple(parseddate)) supported = int(index_tag.section["Supported"]) version = index_tag.section["Version"] # add the information to a new date object dist = Dist(name, version, date, supported) if "ReleaseNotes" in index_tag.section: dist.releaseNotesURI = index_tag.section["ReleaseNotes"] lang = get_lang() if lang: dist.releaseNotesURI += "?lang=%s" % lang if "ReleaseNotesHtml" in index_tag.section: dist.releaseNotesHtmlUri = index_tag.section[ "ReleaseNotesHtml"] query = self._get_release_notes_uri_query_string(dist) if query: dist.releaseNotesHtmlUri += query if "UpgradeTool" in index_tag.section: dist.upgradeTool = index_tag.section["UpgradeTool"] if "UpgradeToolSignature" in index_tag.section: dist.upgradeToolSig = index_tag.section[ "UpgradeToolSignature"] if "UpgradeBroken" in index_tag.section: dist.upgrade_broken = index_tag.section["UpgradeBroken"] dists.append(dist) if name == current_dist_name: current_dist = dist except apt_pkg.Error: raise MetaReleaseParseError("Unable to parse %s" % self.METARELEASE_URI) self.metarelease_information.close() self.metarelease_information = None # first check if the current runing distro is in the meta-release # information. if not, we assume that we run on something not # supported and silently return if current_dist is None: self._debug("current dist not found in meta-release file\n") return False # then see what we can upgrade to upgradable_to = "" for dist in dists: if dist.date > current_dist.date: # Only offer to upgrade to an unsupported release if running # with useDevelopmentRelease, this way one can upgrade from an # LTS release to the next supported non-LTS release e.g. from # 14.04 to 15.04. if not dist.supported and not self.useDevelopmentRelease: continue upgradable_to = dist self._debug("new dist: %s" % upgradable_to) break # only warn if unsupported and a new dist is available (because # the development version is also unsupported) if upgradable_to != "" and not current_dist.supported: self.upgradable_to = upgradable_to self.dist_no_longer_supported(current_dist) if upgradable_to != "": self.upgradable_to = upgradable_to self.new_dist_available(upgradable_to) # parsing done and sucessfully return True # the network thread that tries to fetch the meta-index file # can't touch the gui, runs as a thread def download(self): self._debug("MetaRelease.download()") lastmodified = 0 req = Request(self.METARELEASE_URI) # make sure that we always get the latest file (#107716) req.add_header("Cache-Control", "No-Cache") req.add_header("Pragma", "no-cache") if os.access(self.METARELEASE_FILE, os.W_OK): try: lastmodified = os.stat(self.METARELEASE_FILE).st_mtime except OSError: pass if lastmodified > 0 and not self.forceDownload: req.add_header("If-Modified-Since", time.asctime(time.gmtime(lastmodified))) try: # open uri = urlopen(req, timeout=20) # sometime there is a root owned meta-relase file # there, try to remove it so that we get it # with proper permissions if (os.path.exists(self.METARELEASE_FILE) and not os.access(self.METARELEASE_FILE, os.W_OK)): try: os.unlink(self.METARELEASE_FILE) except OSError as e: print("Can't unlink '%s' (%s)" % (self.METARELEASE_FILE, e)) # we may get exception here on e.g. disk full try: f = open(self.METARELEASE_FILE, "w+") for line in uri.readlines(): f.write(line.decode("UTF-8")) f.flush() f.seek(0, 0) self.metarelease_information = f except IOError: pass uri.close() # http error except HTTPError as e: # mvo: only reuse local info on "not-modified" if e.code == 304 and os.path.exists(self.METARELEASE_FILE): self._debug("reading file '%s'" % self.METARELEASE_FILE) self.metarelease_information = open(self.METARELEASE_FILE, "r") else: self._debug("result of meta-release download: '%s'" % e) # generic network error except (URLError, BadStatusLine, socket.timeout) as e: self._debug("result of meta-release download: '%s'" % e) print("Failed to connect to %s. Check your Internet connection " "or proxy settings" % self.METARELEASE_URI) # now check the information we have if self.metarelease_information is not None: self._debug("have self.metarelease_information") try: self.parse() except Exception: logging.exception("parse failed for '%s'" % self.METARELEASE_FILE) # no use keeping a broken file around os.remove(self.METARELEASE_FILE) # we don't want to keep a meta-release file around when it # has a "Broken" flag, this ensures we are not bitten by # I-M-S/cache issues if self.new_dist and self.new_dist.upgrade_broken: os.remove(self.METARELEASE_FILE) else: self._debug("NO self.metarelease_information") self.downloaded.set() @property def downloading(self): return not self.downloaded.is_set() def _get_release_notes_uri_query_string(self, dist): q = "?" # get the lang lang = get_lang() if lang: q += "lang=%s&" % lang # get the os q += "os=%s&" % self.flavor # get the version to upgrade to q += "ver=%s" % dist.version # the archive didn't respond well to ? being %3F return quote(q, '/?') def _debug(self, msg): if self.DEBUG: sys.stderr.write(msg + "\n") if __name__ == "__main__": meta = MetaReleaseCore(False, False) MyCache.py 0000644 00000044705 15030661120 0006433 0 ustar 00 # MyCache.py # -*- Mode: Python; indent-tabs-mode: nil; tab-width: 4; coding: utf-8 -*- # # Copyright (c) 2004-2008 Canonical # # Author: Michael Vogt <mvo@debian.org> # # This program is free software; you can redistribute it and/or # modify it under the terms of the GNU General Public License as # published by the Free Software Foundation; either version 2 of the # License, or (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program; if not, write to the Free Software # Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 # USA from __future__ import absolute_import, print_function import warnings warnings.filterwarnings("ignore", "apt API not stable yet", FutureWarning) import apt import apt_pkg import logging import os try: from urllib.error import HTTPError from urllib.request import urlopen from urllib.parse import urlsplit except ImportError: from urllib2 import HTTPError, urlopen from urlparse import urlsplit try: from http.client import BadStatusLine except ImportError: from httplib import BadStatusLine import socket import subprocess import re import DistUpgrade.DistUpgradeCache from gettext import gettext as _ try: from launchpadlib.launchpad import Launchpad except ImportError: Launchpad = None SYNAPTIC_PINFILE = "/var/lib/synaptic/preferences" CHANGELOGS_POOL = "https://changelogs.ubuntu.com/changelogs/pool/" CHANGELOGS_URI = CHANGELOGS_POOL + "%s/%s/%s/%s_%s/%s" class HttpsChangelogsUnsupportedError(Exception): """ https changelogs with credentials are unsupported because of the lack of certitifcation validation in urllib2 which allows MITM attacks to steal the credentials """ pass class MyCache(DistUpgrade.DistUpgradeCache.MyCache): CHANGELOG_ORIGIN = "Ubuntu" def __init__(self, progress, rootdir=None): apt.Cache.__init__(self, progress, rootdir) # save for later self.rootdir = rootdir # raise if we have packages in reqreinst state # and let the caller deal with that (runs partial upgrade) assert len(self.req_reinstall_pkgs) == 0 # check if the dpkg journal is ok (we need to do that here # too because libapt will only do it when it tries to lock # the packaging system) assert(not self._dpkgJournalDirty()) # init the regular cache self._initDepCache() self.all_changes = {} self.all_news = {} self.pro_versions = {} # on broken packages, try to fix via saveDistUpgrade() if self._depcache.broken_count > 0: self.saveDistUpgrade() assert (self._depcache.broken_count == 0 and self._depcache.del_count == 0) self.launchpad = None # generate versioned_kernel_pkgs_regexp for later use apt_versioned_kernel_pkgs = apt_pkg.config.value_list( "APT::VersionedKernelPackages") if apt_versioned_kernel_pkgs: self.versioned_kernel_pkgs_regexp = re.compile("(" + "|".join( ["^" + p for p in apt_versioned_kernel_pkgs]) + ")") running_kernel_version = subprocess.check_output( ["uname", "-r"], universal_newlines=True).rstrip() self.running_kernel_pkgs_regexp = re.compile("(" + "|".join( [("^" + p + ".*" + running_kernel_version) if not p.startswith(".*") else (running_kernel_version + p) for p in apt_versioned_kernel_pkgs]) + ")") else: self.versioned_kernel_pkgs_regexp = None self.running_kernel_pkgs_regexp = None def _dpkgJournalDirty(self): """ test if the dpkg journal is dirty (similar to debSystem::CheckUpdates) """ d = os.path.dirname( apt_pkg.config.find_file("Dir::State::status")) + "/updates" for f in os.listdir(d): if re.match("[0-9]+", f): return True return False def _initDepCache(self): #apt_pkg.config.set("Debug::pkgPolicy","1") #self.depcache = apt_pkg.GetDepCache(self.cache) #self._depcache = apt_pkg.GetDepCache(self._cache) self._depcache.read_pinfile() if os.path.exists(SYNAPTIC_PINFILE): self._depcache.read_pinfile(SYNAPTIC_PINFILE) self._depcache.init() def clear(self): self._initDepCache() @property def required_download(self): """ get the size of the packages that are required to download """ pm = apt_pkg.PackageManager(self._depcache) fetcher = apt_pkg.Acquire() pm.get_archives(fetcher, self._list, self._records) return fetcher.fetch_needed @property def install_count(self): return self._depcache.inst_count def keep_count(self): return self._depcache.keep_count @property def del_count(self): return self._depcache.del_count def _check_dependencies(self, target, deps): """Return True if any of the dependencies in deps match target.""" # TODO: handle virtual packages for dep_or in deps: if not dep_or: continue match = True for base_dep in dep_or: if (base_dep.name != target.package.shortname or not apt_pkg.check_dep( target.version, base_dep.relation, base_dep.version)): match = False if match: return True return False def find_removal_justification(self, pkg): target = pkg.installed if not target: return False for cpkg in self: candidate = cpkg.candidate if candidate is not None: if (self._check_dependencies( target, candidate.get_dependencies("Conflicts")) and self._check_dependencies( target, candidate.get_dependencies("Replaces"))): logging.info( "%s Conflicts/Replaces %s; allowing removal" % ( candidate.package.shortname, pkg.shortname)) return True return False def saveDistUpgrade(self): """ this functions mimics a upgrade but will never remove anything """ #self._apply_dselect_upgrade() self._depcache.upgrade(True) wouldDelete = self._depcache.del_count if wouldDelete > 0: deleted_pkgs = [pkg for pkg in self if pkg.marked_delete] assert wouldDelete == len(deleted_pkgs) for pkg in deleted_pkgs: if self.find_removal_justification(pkg): wouldDelete -= 1 if wouldDelete > 0: self.clear() assert (self._depcache.broken_count == 0 and self._depcache.del_count == 0) else: assert self._depcache.broken_count == 0 #self._apply_dselect_upgrade() self._depcache.upgrade() return wouldDelete def _strip_epoch(self, verstr): " strip of the epoch " vers_no_epoch = verstr.split(":") if len(vers_no_epoch) > 1: verstr = "".join(vers_no_epoch[1:]) return verstr def _get_changelog_or_news(self, name, fname, strict_versioning=False, changelogs_uri=None): " helper that fetches the file in question " # don't touch the gui in this function, it needs to be thread-safe pkg = self[name] # get the src package name srcpkg = pkg.candidate.source_name # assume "main" section src_section = "main" # use the section of the candidate as a starting point section = pkg._pcache._depcache.get_candidate_ver(pkg._pkg).section # get the source version srcver_epoch = pkg.candidate.source_version srcver = self._strip_epoch(srcver_epoch) split_section = section.split("/") if len(split_section) > 1: src_section = split_section[0] # lib is handled special prefix = srcpkg[0] if srcpkg.startswith("lib"): prefix = "lib" + srcpkg[3] # the changelogs_uri argument overrides the default changelogs_uri, # this is useful for e.g. PPAs where we construct the changelogs # path differently if changelogs_uri: uri = changelogs_uri else: uri = CHANGELOGS_URI % (src_section, prefix, srcpkg, srcpkg, srcver, fname) # https uris are not supported when they contain a username/password # because the urllib2 https implementation will not check certificates # and so its possible to do a man-in-the-middle attack to steal the # credentials res = urlsplit(uri) if res.scheme == "https" and res.username: raise HttpsChangelogsUnsupportedError( "https locations with username/password are not" "supported to fetch changelogs") # print("Trying: %s " % uri) changelog = urlopen(uri) #print(changelog.read()) # do only get the lines that are new alllines = "" regexp = "^%s \\((.*)\\)(.*)$" % (re.escape(srcpkg)) while True: line = changelog.readline().decode("UTF-8", "replace") if line == "": break match = re.match(regexp, line) if match: # strip epoch from installed version # and from changelog too installed = getattr(pkg.installed, "version", None) if installed and ":" in installed: installed = installed.split(":", 1)[1] changelogver = match.group(1) if changelogver and ":" in changelogver: changelogver = changelogver.split(":", 1)[1] # we test for "==" here for changelogs # to ensure that the version # is actually really in the changelog - if not # just display it all, this catches cases like: # gcc-defaults with "binver=4.3.1" and srcver=1.76 # # for NEWS.Debian we do require the changelogver > installed if strict_versioning: if (installed and apt_pkg.version_compare(changelogver, installed) < 0): break else: if (installed and apt_pkg.version_compare(changelogver, installed) == 0): break alllines = alllines + line return alllines def _extract_ppa_changelog_uri(self, name): """Return the changelog URI from the Launchpad API Return None in case of an error. """ if not Launchpad: logging.warning("Launchpadlib not available, cannot retrieve PPA " "changelog") return None cdt = self[name].candidate for uri in cdt.uris: if urlsplit(uri).hostname != 'ppa.launchpad.net': continue match = re.search('http.*/(.*)/(.*)/ubuntu/.*', uri) if match is not None: user, ppa = match.group(1), match.group(2) break else: logging.error("Unable to find a valid PPA candidate URL.") return # Login on launchpad if we are not already if self.launchpad is None: self.launchpad = Launchpad.login_anonymously('update-manager', 'production', version='devel') archive = self.launchpad.archives.getByReference( reference='~%s/ubuntu/%s' % (user, ppa) ) if archive is None: logging.error("Unable to retrieve the archive from the Launchpad " "API.") return spphs = archive.getPublishedSources(source_name=cdt.source_name, exact_match=True, version=cdt.source_version) if not spphs: logging.error("No published sources were retrieved from the " "Launchpad API.") return return spphs[0].changelogUrl() def _guess_third_party_changelogs_uri_by_source(self, name): pkg = self[name] deb_uri = pkg.candidate.uri if deb_uri is None: return None srcrec = pkg.candidate.record.get("Source") if not srcrec: return None # srcpkg can be "apt" or "gcc-default (1.0)" srcpkg = srcrec.split("(")[0].strip() if "(" in srcrec: srcver = srcrec.split("(")[1].rstrip(")") else: srcver = pkg.candidate.source_version base_uri = deb_uri.rpartition("/")[0] return base_uri + "/%s_%s.changelog" % (srcpkg, srcver) def _guess_third_party_changelogs_uri_by_binary(self, name): """ guess changelogs uri based on ArchiveURI by replacing .deb with .changelog """ # there is always a pkg and a pkg.candidate, no need to add # check here pkg = self[name] deb_uri = pkg.candidate.uri if deb_uri: return "%s.changelog" % deb_uri.rsplit(".", 1)[0] return None def get_news_and_changelog(self, name, lock): self.get_news(name) self.get_changelog(name) try: lock.release() except Exception: pass def get_news(self, name): " get the NEWS.Debian file from the changelogs location " try: news = self._get_changelog_or_news(name, "NEWS.Debian", True) except Exception: return if news: self.all_news[name] = news def _fetch_changelog_for_third_party_package(self, name, origins): # Special case for PPAs changelogs_uri_ppa = None for origin in origins: if origin.origin.startswith('LP-PPA-'): try: changelogs_uri_ppa = self._extract_ppa_changelog_uri(name) break except Exception: logging.exception("Unable to connect to the Launchpad " "API.") # Try non official changelog location changelogs_uri_binary = \ self._guess_third_party_changelogs_uri_by_binary(name) changelogs_uri_source = \ self._guess_third_party_changelogs_uri_by_source(name) error_message = "" for changelogs_uri in [changelogs_uri_ppa, changelogs_uri_binary, changelogs_uri_source]: if changelogs_uri: try: changelog = self._get_changelog_or_news( name, "changelog", False, changelogs_uri) self.all_changes[name] += changelog except (HTTPError, HttpsChangelogsUnsupportedError): # no changelogs_uri or 404 error_message = _( "This update does not come from a " "source that supports changelogs.") except (IOError, BadStatusLine, socket.error): # network errors and others logging.exception("error on changelog fetching") error_message = _( "Failed to download the list of changes. \n" "Please check your Internet connection.") self.all_changes[name] += error_message # If the machine is not attached to Ubuntu Pro, Update Manager advertises # the upgrades that would be available if it were attached. # As that is unbeknownst to Apt, we need this map to show the correct # version of each upgradable-if-pro-subscribed package. def create_pro_cache(self, pro_pkgs): for (name, version, _a, _a) in pro_pkgs: self.pro_versions[name] = version def get_changelog(self, name): " get the changelog file from the changelog location " origins = self[name].candidate.origins version = self.pro_versions.get(name, self[name].candidate.version) self.all_changes[name] = _("Changes for %s versions:\n" "Installed version: %s\n" "Available version: %s\n\n") % \ (name, getattr(self[name].installed, "version", None), version) if self.CHANGELOG_ORIGIN not in [o.origin for o in origins]: self._fetch_changelog_for_third_party_package(name, origins) return # fixup epoch handling version srcpkg = self[name].candidate.source_name srcver_epoch = self[name].candidate.source_version.replace(':', '%3A') try: changelog = self._get_changelog_or_news(name, "changelog") if len(changelog) == 0: changelog = _("The changelog does not contain any relevant " "changes.\n\n" "Please use http://launchpad.net/ubuntu/+source/" "%s/%s/+changelog\n" "until the changes become available or try " "again later.") % (srcpkg, srcver_epoch) except HTTPError: changelog = _("The list of changes is not available yet.\n\n" "Please use http://launchpad.net/ubuntu/+source/" "%s/%s/+changelog\n" "until the changes become available or try again " "later.") % (srcpkg, srcver_epoch) except (IOError, BadStatusLine, socket.error) as e: print("caught exception: ", e) changelog = _("Failed to download the list " "of changes. \nPlease " "check your Internet " "connection.") self.all_changes[name] += changelog roam.py 0000644 00000017772 15030661120 0006064 0 ustar 00 # utils.py # -*- Mode: Python; indent-tabs-mode: nil; tab-width: 4; coding: utf-8 -*- # # Copyright (c) 2011 Canonical # # Author: Alex Chiang <achiang@canonical.com> # Michael Vogt <michael.vogt@ubuntu.com> # # This program is free software; you can redistribute it and/or # modify it under the terms of the GNU General Public License as # published by the Free Software Foundation; either version 2 of the # License, or (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program; if not, write to the Free Software # Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 # USA from __future__ import print_function import dbus import sys class ModemManagerHelper(object): # data taken from # http://projects.gnome.org/NetworkManager/developers/mm-spec-04.html MM_DBUS_IFACE = "org.freedesktop.ModemManager" MM_DBUS_IFACE_MODEM = MM_DBUS_IFACE + ".Modem" # MM_MODEM_TYPE MM_MODEM_TYPE_GSM = 1 MM_MODEM_TYPE_CDMA = 2 # GSM # Not registered, not searching for new operator to register. MM_MODEM_GSM_NETWORK_REG_STATUS_IDLE = 0 # Registered on home network. MM_MODEM_GSM_NETWORK_REG_STATUS_HOME = 1 # Not registered, searching for new operator to register with. MM_MODEM_GSM_NETWORK_REG_STATUS_SEARCHING = 2 # Registration denied. MM_MODEM_GSM_NETWORK_REG_STATUS_DENIED = 3 # Unknown registration status. MM_MODEM_GSM_NETWORK_REG_STATUS_UNKNOWN = 4 # Registered on a roaming network. MM_MODEM_GSM_NETWORK_REG_STATUS_ROAMING = 5 # CDMA # Registration status is unknown or the device is not registered. MM_MODEM_CDMA_REGISTRATION_STATE_UNKNOWN = 0 # Registered, but roaming status is unknown or cannot be provided # by the device. The device may or may not be roaming. MM_MODEM_CDMA_REGISTRATION_STATE_REGISTERED = 1 # Currently registered on the home network. MM_MODEM_CDMA_REGISTRATION_STATE_HOME = 2 # Currently registered on a roaming network. MM_MODEM_CDMA_REGISTRATION_STATE_ROAMING = 3 def __init__(self): self.bus = dbus.SystemBus() self.proxy = self.bus.get_object("org.freedesktop.ModemManager", "/org/freedesktop/ModemManager") modem_manager = dbus.Interface(self.proxy, self.MM_DBUS_IFACE) self.modems = modem_manager.EnumerateDevices() @staticmethod def get_dbus_property(proxy, interface, property): props = dbus.Interface(proxy, "org.freedesktop.DBus.Properties") property = props.Get(interface, property) return property def is_gsm_roaming(self): for m in self.modems: dev = self.bus.get_object(self.MM_DBUS_IFACE, m) type = self.get_dbus_property(dev, self.MM_DBUS_IFACE_MODEM, "Type") if type != self.MM_MODEM_TYPE_GSM: continue net = dbus.Interface(dev, self.MM_DBUS_IFACE_MODEM + ".Gsm.Network") reg = net.GetRegistrationInfo() # Be conservative about roaming. If registration unknown, # assume yes. # MM_MODEM_GSM_NETWORK_REG_STATUS if reg[0] in (self.MM_MODEM_GSM_NETWORK_REG_STATUS_UNKNOWN, self.MM_MODEM_GSM_NETWORK_REG_STATUS_ROAMING): return True return False def is_cdma_roaming(self): for m in self.modems: dev = self.bus.get_object(self.MM_DBUS_IFACE, m) type = self.get_dbus_property(dev, self.MM_DBUS_IFACE_MODEM, "Type") if type != self.MM_MODEM_TYPE_CDMA: continue cdma = dbus.Interface(dev, self.MM_DBUS_IFACE_MODEM + ".Cdma") (cmda_1x, evdo) = cdma.GetRegistrationState() # Be conservative about roaming. If registration unknown, # assume yes. # MM_MODEM_CDMA_REGISTRATION_STATE roaming_states = (self.MM_MODEM_CDMA_REGISTRATION_STATE_REGISTERED, self.MM_MODEM_CDMA_REGISTRATION_STATE_ROAMING) # evdo trumps cmda_1x (thanks to Mathieu Trudel-Lapierre) if evdo in roaming_states: return True elif cmda_1x in roaming_states: return True return False class NetworkManagerHelper(object): NM_DBUS_IFACE = "org.freedesktop.NetworkManager" # connection states # Old enum values are for NM 0.7 # The NetworkManager daemon is in an unknown state. NM_STATE_UNKNOWN = 0 # The NetworkManager daemon is connecting a device. NM_STATE_CONNECTING_OLD = 2 NM_STATE_CONNECTING = 40 NM_STATE_CONNECTING_LIST = [NM_STATE_CONNECTING_OLD, NM_STATE_CONNECTING] # The NetworkManager daemon is connected. NM_STATE_CONNECTED_OLD = 3 NM_STATE_CONNECTED_LOCAL = 50 NM_STATE_CONNECTED_SITE = 60 NM_STATE_CONNECTED_GLOBAL = 70 NM_STATE_CONNECTED_LIST = [NM_STATE_CONNECTED_OLD, NM_STATE_CONNECTED_LOCAL, NM_STATE_CONNECTED_SITE, NM_STATE_CONNECTED_GLOBAL] # The device type is unknown. NM_DEVICE_TYPE_UNKNOWN = 0 # The device is wired Ethernet device. NM_DEVICE_TYPE_ETHERNET = 1 # The device is an 802.11 WiFi device. NM_DEVICE_TYPE_WIFI = 2 # The device is a GSM-based cellular WAN device. NM_DEVICE_TYPE_GSM = 3 # The device is a CDMA/IS-95-based cellular WAN device. NM_DEVICE_TYPE_CDMA = 4 def __init__(self): self.bus = dbus.SystemBus() self.proxy = self.bus.get_object("org.freedesktop.NetworkManager", "/org/freedesktop/NetworkManager") @staticmethod def get_dbus_property(proxy, interface, property): props = dbus.Interface(proxy, "org.freedesktop.DBus.Properties") property = props.Get(interface, property) return property def is_active_connection_gsm_or_cdma(self): res = False actives = self.get_dbus_property( self.proxy, self.NM_DBUS_IFACE, 'ActiveConnections') for a in actives: active = self.bus.get_object(self.NM_DBUS_IFACE, a) default_route = self.get_dbus_property( active, self.NM_DBUS_IFACE + ".Connection.Active", 'Default') if not default_route: continue devs = self.get_dbus_property( active, self.NM_DBUS_IFACE + ".Connection.Active", 'Devices') for d in devs: dev = self.bus.get_object(self.NM_DBUS_IFACE, d) type = self.get_dbus_property( dev, self.NM_DBUS_IFACE + ".Device", 'DeviceType') if type == self.NM_DEVICE_TYPE_GSM: return True elif type == self.NM_DEVICE_TYPE_CDMA: return True else: continue return res def is_active_connection_gsm_or_cdma_roaming(self): res = False if self.is_active_connection_gsm_or_cdma(): mmhelper = ModemManagerHelper() res |= mmhelper.is_gsm_roaming() res |= mmhelper.is_cdma_roaming() return res if __name__ == "__main__": # test code if sys.argv[1:] and sys.argv[1] == "--test": mmhelper = ModemManagerHelper() print("is_gsm_roaming", mmhelper.is_gsm_roaming()) print("is_cdma_romaing", mmhelper.is_cdma_roaming()) # roaming? nmhelper = NetworkManagerHelper() is_roaming = nmhelper.is_active_connection_gsm_or_cdma_roaming() print("roam: ", is_roaming) if is_roaming: sys.exit(1) sys.exit(0) LivePatchSocket.py 0000644 00000007322 15030661120 0010144 0 ustar 00 # LivePatchSocket.py # -*- Mode: Python; indent-tabs-mode: nil; tab-width: 4; coding: utf-8 -*- # # Copyright (c) 2017 Canonical # # Author: Andrea Azzarone <andrea.azzarone@canonical.com> # # This program is free software; you can redistribute it and/or # modify it under the terms of the GNU General Public License as # published by the Free Software Foundation; either version 2 of the # License, or (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program; if not, write to the Free Software # Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 # USA from gi.repository import GLib import http.client import socket import threading import yaml HOST_NAME = '/var/snap/canonical-livepatch/current/livepatchd.sock' class UHTTPConnection(http.client.HTTPConnection): def __init__(self, path): http.client.HTTPConnection.__init__(self, 'localhost') self.path = path def connect(self): sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) sock.connect(self.path) self.sock = sock class LivePatchSocket(object): def __init__(self, http_conn=None): if http_conn is None: self.conn = UHTTPConnection(HOST_NAME) else: self.conn = http_conn def get_status(self, on_done): def do_call(): try: self.conn.request('GET', '/status?verbose=True') r = self.conn.getresponse() active = r.status == 200 data = yaml.safe_load(r.read()) except Exception: active = False data = dict() check_state = LivePatchSocket.get_check_state(data) patch_state = LivePatchSocket.get_patch_state(data) fixes = LivePatchSocket.get_fixes(data) GLib.idle_add(lambda: on_done( active, check_state, patch_state, fixes)) thread = threading.Thread(target=do_call) thread.start() @staticmethod def get_check_state(data): try: status = data['status'] kernel = next((k for k in status if k['running']), None) return kernel['livepatch']['checkState'] except Exception: return 'check-failed' @staticmethod def get_patch_state(data): try: status = data['status'] kernel = next((k for k in status if k['running']), None) return kernel['livepatch']['patchState'] except Exception: return 'unknown' @staticmethod def get_fixes(data): try: status = data['status'] kernel = next((k for k in status if k['running']), None) fixes = kernel['livepatch']['fixes'] return [LivePatchFix(f) for f in fixes.replace('* ', '').split('\n') if len(f) > 0] except Exception: return list() class LivePatchFix(object): def __init__(self, text): patched_pattern = ' (unpatched)' self.patched = text.find(patched_pattern) == -1 self.name = text.replace(patched_pattern, '') def __eq__(self, other): if isinstance(other, LivePatchFix): return self.name == other.name and self.patched == other.patched return NotImplemented def __ne__(self, other): result = self.__eq__(other) if result is NotImplemented: return result return not result AlertWatcher.py 0000644 00000007765 15030661120 0007514 0 ustar 00 # AlertWatcher.py # -*- Mode: Python; indent-tabs-mode: nil; tab-width: 4; coding: utf-8 -*- # # Copyright (c) 2010 Mohamed Amine IL Idrissi # # Author: Mohamed Amine IL Idrissi <ilidrissiamine@gmail.com> # # This program is free software; you can redistribute it and/or # modify it under the terms of the GNU General Public License as # published by the Free Software Foundation; either version 2 of the # License, or (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program; if not, write to the Free Software # Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 # USA from __future__ import absolute_import from gi.repository import GObject import dbus from dbus.mainloop.glib import DBusGMainLoop class AlertWatcher(GObject.GObject): """ a class that checks for alerts and reports them, like a battery or network warning """ __gsignals__ = {"network-alert": (GObject.SignalFlags.RUN_FIRST, None, (GObject.TYPE_INT,)), "battery-alert": (GObject.SignalFlags.RUN_FIRST, None, (GObject.TYPE_BOOLEAN,)), "network-3g-alert": (GObject.SignalFlags.RUN_FIRST, None, (GObject.TYPE_BOOLEAN, GObject.TYPE_BOOLEAN,)), } def __init__(self): GObject.GObject.__init__(self) DBusGMainLoop(set_as_default=True) self.bus = dbus.Bus(dbus.Bus.TYPE_SYSTEM) # make it always connected if NM isn't available self.network_state = 3 def check_alert_state(self): try: obj = self.bus.get_object("org.freedesktop.NetworkManager", "/org/freedesktop/NetworkManager") obj.connect_to_signal( "StateChanged", self._on_network_state_changed, dbus_interface="org.freedesktop.NetworkManager") interface = dbus.Interface(obj, "org.freedesktop.DBus.Properties") self.network_state = interface.Get( "org.freedesktop.NetworkManager", "State") self._network_alert(self.network_state) # power obj = self.bus.get_object('org.freedesktop.UPower', '/org/freedesktop/UPower') obj.connect_to_signal("Changed", self._power_changed, dbus_interface="org.freedesktop.UPower") self._power_changed() # 3g self._update_3g_state() except dbus.exceptions.DBusException: pass def _on_network_state_changed(self, state): self._network_alert(state) self._update_3g_state() def _update_3g_state(self): from .roam import NetworkManagerHelper nm = NetworkManagerHelper() on_3g = nm.is_active_connection_gsm_or_cdma() is_roaming = nm.is_active_connection_gsm_or_cdma_roaming() self._network_3g_alert(on_3g, is_roaming) def _network_3g_alert(self, on_3g, is_roaming): self.emit("network-3g-alert", on_3g, is_roaming) def _network_alert(self, state): self.network_state = state self.emit("network-alert", state) def _power_changed(self): obj = self.bus.get_object("org.freedesktop.UPower", "/org/freedesktop/UPower") interface = dbus.Interface(obj, "org.freedesktop.DBus.Properties") on_battery = interface.Get("org.freedesktop.UPower", "OnBattery") self.emit("battery-alert", on_battery) __init__.py 0000644 00000000000 15030661120 0006635 0 ustar 00 utils.py 0000644 00000044155 15030661120 0006261 0 ustar 00 # utils.py # -*- Mode: Python; indent-tabs-mode: nil; tab-width: 4; coding: utf-8 -*- # # Copyright (c) 2004-2013 Canonical # # Authors: Michael Vogt <mvo@debian.org> # Michael Terry <michael.terry@canonical.com> # # This program is free software; you can redistribute it and/or # modify it under the terms of the GNU General Public License as # published by the Free Software Foundation; either version 2 of the # License, or (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program; if not, write to the Free Software # Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 # USA from __future__ import print_function from gettext import gettext as _ from gettext import ngettext from stat import (S_IMODE, ST_MODE, S_IXUSR) from math import ceil import apt import apt_pkg apt_pkg.init_config() import locale import logging import re import os import subprocess import sys import time try: from urllib.request import ( ProxyHandler, Request, build_opener, install_opener, urlopen, ) from urllib.parse import urlsplit except ImportError: from urllib2 import ( ProxyHandler, Request, build_opener, install_opener, urlopen, ) from urlparse import urlsplit from copy import copy class ExecutionTime(object): """ Helper that can be used in with statements to have a simple measure of the timing of a particular block of code, e.g. with ExecutionTime("db flush"): db.flush() """ def __init__(self, info=""): self.info = info def __enter__(self): self.now = time.time() def __exit__(self, type, value, stack): print("%s: %s" % (self.info, time.time() - self.now)) class SoftwarePropertiesPage(): ubuntu_software = 0 other_software = 1 updates = 2 authentication = 3 additional_drivers = 4 developer_options = 5 ubuntu_pro = 6 def get_string_with_no_auth_from_source_entry(entry): tmp = copy(entry) url_parts = urlsplit(tmp.uri) if url_parts.username: tmp.uri = tmp.uri.replace(url_parts.username, "hidden-u") if url_parts.password: tmp.uri = tmp.uri.replace(url_parts.password, "hidden-p") return str(tmp) def is_unity_running(): """ return True if Unity is currently running """ unity_running = False try: import dbus bus = dbus.SessionBus() unity_running = bus.name_has_owner("com.canonical.Unity") except Exception: logging.exception("could not check for Unity dbus service") return unity_running def is_child_of_process_name(processname, pid=None): if not pid: pid = os.getpid() while pid > 0: stat_file = "/proc/%s/stat" % pid with open(stat_file) as stat_f: stat = stat_f.read() # extract command (inside ()) command = stat.partition("(")[2].rpartition(")")[0] if command == processname: return True # get parent (second to the right of command) and check that next pid = int(stat.rpartition(")")[2].split()[1]) return False def inside_chroot(): """ returns True if we are inside a chroot """ # if there is no proc or no pid 1 we are very likely inside a chroot if not os.path.exists("/proc") or not os.path.exists("/proc/1"): return True # if the inode is differnt for pid 1 "/" and our "/" return os.stat("/") != os.stat("/proc/1/root") def wrap(t, width=70, subsequent_indent=""): """ helpers inspired after textwrap - unfortunately we can not use textwrap directly because it break packagenames with "-" in them into new lines """ out = "" for s in t.split(): if (len(out) - out.rfind("\n")) + len(s) > width: out += "\n" + subsequent_indent out += s + " " return out def twrap(s, **kwargs): msg = "" paras = s.split("\n") for par in paras: s = wrap(par, **kwargs) msg += s + "\n" return msg def lsmod(): " return list of loaded modules (or [] if lsmod is not found) " modules = [] # FIXME raise? if not os.path.exists("/sbin/lsmod"): return [] p = subprocess.Popen(["/sbin/lsmod"], stdout=subprocess.PIPE, universal_newlines=True) lines = p.communicate()[0].split("\n") # remove heading line: "Modules Size Used by" del lines[0] # add lines to list, skip empty lines for line in lines: if line: modules.append(line.split()[0]) return modules def check_and_fix_xbit(path): " check if a given binary has the executable bit and if not, add it" if not os.path.exists(path): return mode = S_IMODE(os.stat(path)[ST_MODE]) if not ((mode & S_IXUSR) == S_IXUSR): os.chmod(path, mode | S_IXUSR) def country_mirror(): " helper to get the country mirror from the current locale " # special cases go here lang_mirror = {'c': ''} # no lang, no mirror if 'LANG' not in os.environ: return '' lang = os.environ['LANG'].lower() # check if it is a special case if lang[:5] in lang_mirror: return lang_mirror[lang[:5]] # now check for the most comon form (en_US.UTF-8) if "_" in lang: country = lang.split(".")[0].split("_")[1] if "@" in country: country = country.split("@")[0] return country + "." else: return lang[:2] + "." return '' def get_dist(): " return the codename of the current runing distro " # support debug overwrite dist = os.environ.get("META_RELEASE_FAKE_CODENAME") if dist: logging.warning("using fake release name '%s' (because of " "META_RELEASE_FAKE_CODENAME environment) " % dist) return dist # then check the real one from subprocess import Popen, PIPE p = Popen(["lsb_release", "-c", "-s"], stdout=PIPE, universal_newlines=True) res = p.wait() if res != 0: sys.stderr.write("lsb_release returned exitcode: %i\n" % res) return "unknown distribution" dist = p.stdout.readline().strip() p.stdout.close() return dist def get_dist_version(): " return the version of the current running distro " # support debug overwrite desc = os.environ.get("META_RELEASE_FAKE_VERSION") if desc: logging.warning("using fake release version '%s' (because of " "META_RELEASE_FAKE_VERSION environment) " % desc) return desc # then check the real one from subprocess import Popen, PIPE p = Popen(["lsb_release", "-r", "-s"], stdout=PIPE, universal_newlines=True) res = p.wait() if res != 0: sys.stderr.write("lsb_release returned exitcode: %i\n" % res) return "unknown distribution" desc = p.stdout.readline().strip() p.stdout.close() return desc class HeadRequest(Request): def get_method(self): return "HEAD" def url_downloadable(uri, debug_func=None): """ helper that checks if the given uri exists and is downloadable (supports optional debug_func function handler to support e.g. logging) Supports http (via HEAD) and ftp (via size request) """ if not debug_func: lambda x: True debug_func("url_downloadable: %s" % uri) (scheme, netloc, path, querry, fragment) = urlsplit(uri) debug_func("s='%s' n='%s' p='%s' q='%s' f='%s'" % (scheme, netloc, path, querry, fragment)) if scheme in ("http", "https"): try: http_file = urlopen(HeadRequest(uri)) http_file.close() if http_file.code == 200: return True return False except Exception as e: debug_func("error from httplib: '%s'" % e) return False elif scheme == "ftp": import ftplib try: f = ftplib.FTP(netloc) f.login() f.cwd(os.path.dirname(path)) size = f.size(os.path.basename(path)) f.quit() if debug_func: debug_func("ftplib.size() returned: %s" % size) if size != 0: return True except Exception as e: if debug_func: debug_func("error from ftplib: '%s'" % e) return False return False def init_proxy(gsettings=None): """ init proxy settings * use apt.conf http proxy if present, * otherwise look into synaptics config file, * otherwise the default behavior will use http_proxy environment if present """ SYNAPTIC_CONF_FILE = "/root/.synaptic/synaptic.conf" proxies = {} # generic apt config wins if apt_pkg.config.find("Acquire::http::Proxy") != '': proxies["http"] = apt_pkg.config.find("Acquire::http::Proxy") # then synaptic elif os.path.exists(SYNAPTIC_CONF_FILE): cnf = apt_pkg.Configuration() apt_pkg.read_config_file(cnf, SYNAPTIC_CONF_FILE) use_proxy = cnf.find_b("Synaptic::useProxy", False) if use_proxy: proxy_host = cnf.find("Synaptic::httpProxy") proxy_port = str(cnf.find_i("Synaptic::httpProxyPort")) if proxy_host and proxy_port: proxies["http"] = "http://%s:%s/" % (proxy_host, proxy_port) if apt_pkg.config.find("Acquire::https::Proxy") != '': proxies["https"] = apt_pkg.config.find("Acquire::https::Proxy") elif "http" in proxies: proxies["https"] = proxies["http"] # if we have a proxy, set it if proxies: # basic verification for proxy in proxies.values(): if not re.match("https?://\\w+", proxy): print("proxy '%s' looks invalid" % proxy, file=sys.stderr) return proxy_support = ProxyHandler(proxies) opener = build_opener(proxy_support) install_opener(opener) if "http" in proxies: os.putenv("http_proxy", proxies["http"]) if "https" in proxies: os.putenv("https_proxy", proxies["https"]) return proxies def on_battery(): """ Check via dbus if the system is running on battery. This function is using UPower per default, if UPower is not available it falls-back to DeviceKit.Power. """ try: import dbus bus = dbus.Bus(dbus.Bus.TYPE_SYSTEM) try: devobj = bus.get_object('org.freedesktop.UPower', '/org/freedesktop/UPower') dev = dbus.Interface(devobj, 'org.freedesktop.DBus.Properties') return dev.Get('org.freedesktop.UPower', 'OnBattery') except dbus.exceptions.DBusException as e: error_unknown = 'org.freedesktop.DBus.Error.ServiceUnknown' if e._dbus_error_name != error_unknown: raise devobj = bus.get_object('org.freedesktop.DeviceKit.Power', '/org/freedesktop/DeviceKit/Power') dev = dbus.Interface(devobj, "org.freedesktop.DBus.Properties") return dev.Get("org.freedesktop.DeviceKit.Power", "on_battery") except Exception: #import sys #print("on_battery returned error: ", e, file=sys.stderr) return False def inhibit_sleep(): """ Send a dbus signal to logind to not suspend the system, it will be released when the return value drops out of scope """ try: from gi.repository import Gio, GLib connection = Gio.bus_get_sync(Gio.BusType.SYSTEM) var, fdlist = connection.call_with_unix_fd_list_sync( 'org.freedesktop.login1', '/org/freedesktop/login1', 'org.freedesktop.login1.Manager', 'Inhibit', GLib.Variant('(ssss)', ('shutdown:sleep', 'UpdateManager', 'Updating System', 'block')), None, 0, -1, None, None) inhibitor = Gio.UnixInputStream(fd=fdlist.steal_fds()[var[0]]) return inhibitor except Exception: #print("could not send the dbus Inhibit signal: %s" % e) return False def str_to_bool(str): if str == "0" or str.upper() == "FALSE": return False return True def get_lang(): import logging try: (locale_s, encoding) = locale.getdefaultlocale() return locale_s except Exception: logging.exception("gedefaultlocale() failed") return None def get_ubuntu_flavor(cache=None): """ try to guess the flavor based on the running desktop """ # this will (of course) not work in a server environment, # but the main use case for this is to show the right # release notes. pkg = get_ubuntu_flavor_package(cache=cache) return pkg.split('-', 1)[0] def _load_meta_pkg_list(): # This could potentially introduce a circular dependency, but the config # parser logic is simple, and doesn't rely on any UpdateManager code. from DistUpgrade.DistUpgradeConfigParser import DistUpgradeConfig parser = DistUpgradeConfig('/usr/share/ubuntu-release-upgrader') return parser.getlist('Distro', 'MetaPkgs') def get_ubuntu_flavor_package(cache=None): """ try to guess the flavor metapackage based on the running desktop """ # From spec, first if ubuntu-desktop is installed, use that. # Second, grab first installed one from DistUpgrade.cfg. # Lastly, fallback to ubuntu-desktop again. meta_pkgs = ['ubuntu-desktop'] try: meta_pkgs.extend(sorted(_load_meta_pkg_list())) except Exception as e: print('Could not load list of meta packages:', e) if cache is None: cache = apt.Cache() for meta_pkg in meta_pkgs: cache_pkg = cache[meta_pkg] if meta_pkg in cache else None if cache_pkg and cache_pkg.is_installed: return meta_pkg return 'ubuntu-desktop' def get_ubuntu_flavor_name(cache=None): """ try to guess the flavor name based on the running desktop """ pkg = get_ubuntu_flavor_package(cache=cache) lookup = {'ubuntustudio-desktop': 'Ubuntu Studio'} if pkg in lookup: return lookup[pkg] elif pkg.endswith('-desktop'): return capitalize_first_word(pkg.rsplit('-desktop', 1)[0]) elif pkg.endswith('-netbook'): return capitalize_first_word(pkg.rsplit('-netbook', 1)[0]) else: return 'Ubuntu' # Unused by update-manager, but still used by ubuntu-release-upgrader def error(parent, summary, message): import gi gi.require_version("Gtk", "3.0") from gi.repository import Gtk, Gdk d = Gtk.MessageDialog(parent=parent, flags=Gtk.DialogFlags.MODAL, type=Gtk.MessageType.ERROR, buttons=Gtk.ButtonsType.CLOSE) d.set_markup("<big><b>%s</b></big>\n\n%s" % (summary, message)) d.realize() d.get_window().set_functions(Gdk.WMFunction.MOVE) d.set_title("") d.run() d.destroy() return False def humanize_size(bytes): """ Convert a given size in bytes to a nicer better readable unit """ if bytes < 1000 * 1000: # to have 0 for 0 bytes, 1 for 0-1000 bytes and for 1 and above # round up size_in_kb = int(ceil(bytes / float(1000))) # TRANSLATORS: download size of small updates, e.g. "250 kB" return ngettext("%(size).0f kB", "%(size).0f kB", size_in_kb) % { "size": size_in_kb} else: # TRANSLATORS: download size of updates, e.g. "2.3 MB" return locale.format_string(_("%.1f MB"), bytes / 1000.0 / 1000.0) def get_arch(): return apt_pkg.config.find("APT::Architecture") def is_port_already_listening(port): """ check if the current system is listening on the given tcp port """ # index in the line INDEX_LOCAL_ADDR = 1 #INDEX_REMOTE_ADDR = 2 INDEX_STATE = 3 # state (st) that we care about STATE_LISTENING = '0A' # read the data with open("/proc/net/tcp") as net_tcp: for line in net_tcp.readlines(): line = line.strip() if not line: continue # split, values are: # sl local_address rem_address st tx_queue rx_queue tr # tm->when retrnsmt uid timeout inode values = line.split() state = values[INDEX_STATE] if state != STATE_LISTENING: continue local_port_str = values[INDEX_LOCAL_ADDR].split(":")[1] local_port = int(local_port_str, 16) if local_port == port: return True return False def iptables_active(): """ Return True if iptables is active """ # FIXME: is there a better way? iptables_empty = """Chain INPUT (policy ACCEPT) target prot opt source destination Chain FORWARD (policy ACCEPT) target prot opt source destination Chain OUTPUT (policy ACCEPT) target prot opt source destination """ if os.getuid() != 0: raise OSError("Need root to check the iptables state") if not os.path.exists("/sbin/iptables"): return False out = subprocess.Popen(["iptables", "-nL"], stdout=subprocess.PIPE, universal_newlines=True).communicate()[0] if out == iptables_empty: return False return True def capitalize_first_word(string): """ this uppercases the first word's first letter """ if len(string) > 1 and string[0].isalpha() and not string[0].isupper(): return string[0].capitalize() + string[1:] return string def get_package_label(pkg): """ this takes a package synopsis and uppercases the first word's first letter """ name = getattr(pkg.candidate, "summary", "") return capitalize_first_word(name) if __name__ == "__main__": #print(mirror_from_sources_list()) #print(on_battery()) #print(inside_chroot()) #print(iptables_active()) error(None, "bar", "baz") UpdateList.py 0000644 00000060126 15030661120 0007173 0 ustar 00 # UpdateList.py # -*- Mode: Python; indent-tabs-mode: nil; tab-width: 4; coding: utf-8 -*- # # Copyright (c) 2004-2013 Canonical # # Author: Michael Vogt <mvo@debian.org> # Dylan McCall <dylanmccall@ubuntu.com> # Michael Terry <michael.terry@canonical.com> # # This program is free software; you can redistribute it and/or # modify it under the terms of the GNU General Public License as # published by the Free Software Foundation; either version 2 of the # License, or (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program; if not, write to the Free Software # Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 # USA from __future__ import print_function import warnings warnings.filterwarnings("ignore", "Accessed deprecated property", DeprecationWarning) from gettext import gettext as _ import apt import fnmatch import logging import itertools import os import random import glob import re from gi.repository import Gio from UpdateManager.Core import utils class UpdateItem(): def __init__(self, pkg, name, icon, to_remove, sensitive=True): self.icon = icon self.name = name self.pkg = pkg self.to_remove = to_remove self.sensitive = sensitive def is_selected(self): if not self.to_remove: return self.pkg.marked_install or self.pkg.marked_upgrade else: return self.pkg.marked_delete class UpdateGroup(UpdateItem): _depcache = {} def __init__(self, pkg, name, icon, to_remove, sensitive=True): UpdateItem.__init__(self, pkg, name, icon, to_remove, sensitive) self._items = set() self._deps = set() self.core_item = None if pkg is not None: self.core_item = UpdateItem(pkg, name, icon, to_remove, sensitive) self._items.add(self.core_item) @property def items(self): all_items = [] all_items.extend(self._items) return sorted(all_items, key=lambda a: a.name.lower()) def add(self, pkg, cache=None, eventloop_callback=None, to_remove=False, sensitive=True): name = utils.get_package_label(pkg) icon = Gio.ThemedIcon.new("package") self._items.add(UpdateItem(pkg, name, icon, to_remove, sensitive)) # If the pkg is in self._deps, stop here. We have already calculated # the recursive dependencies for this package, no need to do it again. if cache and pkg.name in cache and pkg.name not in self._deps: if not self._deps: # Initial deps haven't been calculated. As we're checking # whether _deps is empty in is_dependency, we must init now or # it won't be done at all. self._init_deps(cache, eventloop_callback) self._add_deps(pkg, cache, eventloop_callback) def contains(self, item): return item in self._items def _init_deps(self, cache, eventloop_callback): for item in self._items: if item.pkg and item.pkg.name not in self._deps: self._add_deps(item.pkg, cache, eventloop_callback) def _add_deps(self, pkg, cache, eventloop_callback): """Adds pkg and dependencies of pkg to the dependency list.""" if pkg is None or pkg.candidate is None or pkg.name in self._deps: # This shouldn't really happen. If we land here often, it's a sign # that something has gone wrong. Unless all pkgs are None it's not # a critical issue - a hit to the performance at most. reason = ((not pkg or not pkg.candidate) and "Package was None or didn't have a candidate." or "%s already in _deps." % pkg.name) logging.debug("Useless call to _add_deps. %s" % reason) return if len(self._deps) % 200 == 0 and callable(eventloop_callback): # Don't spin the loop every time _add_deps is called. eventloop_callback() self._deps.add(pkg.name) if pkg.name in self._depcache: for dep in self._depcache[pkg.name]: if dep not in self._deps and dep in cache: self._add_deps(cache[dep], cache, eventloop_callback) else: candidate = pkg.candidate dependencies = candidate.get_dependencies('Depends', 'Recommends') for dependency_pkg in itertools.chain.from_iterable(dependencies): name = dependency_pkg.name if name not in self._deps and name in cache: self._depcache.setdefault(pkg.name, []).append(name) self._add_deps(cache[name], cache, eventloop_callback) def is_dependency(self, maybe_dep, cache=None, eventloop_callback=None): if not self._deps and cache: self._init_deps(cache, eventloop_callback) return maybe_dep.name in self._deps def packages_are_selected(self): for item in self.items: if item.is_selected(): return True return False def selection_is_inconsistent(self): pkgs_installing = [item for item in self.items if item.is_selected()] return (len(pkgs_installing) > 0 and len(pkgs_installing) < len(self.items)) def get_total_size(self): if self.to_remove: return 0 size = 0 for item in self.items: size += getattr(item.pkg.candidate, "size", 0) return size class UpdateApplicationGroup(UpdateGroup): def __init__(self, pkg, application, to_remove, sensitive=True): name = application.get_display_name() icon = application.get_icon() super(UpdateApplicationGroup, self).__init__(pkg, name, icon, to_remove, sensitive) class UpdatePackageGroup(UpdateGroup): def __init__(self, pkg, to_remove, sensitive=True): name = utils.get_package_label(pkg) icon = Gio.ThemedIcon.new("package") super(UpdatePackageGroup, self).__init__(pkg, name, icon, to_remove, sensitive) class UpdateSystemGroup(UpdateGroup): def __init__(self, cache, to_remove, sensitive=True): # Translators: the %s is a distro name, like 'Ubuntu' and 'base' as in # the core components and packages. name = _("%s base") % utils.get_ubuntu_flavor_name(cache=cache) icon = Gio.ThemedIcon.new("distributor-logo") super(UpdateSystemGroup, self).__init__(None, name, icon, to_remove, sensitive) class UpdateOrigin(): def __init__(self, desc, importance): self.packages = [] self.importance = importance self.description = desc class UpdateList(): """ class that contains the list of available updates in self.pkgs[origin] where origin is the user readable string """ # the key in the debian/control file used to add the phased # updates percentage PHASED_UPDATES_KEY = "Phased-Update-Percentage" # the file that contains the uniq machine id UNIQ_MACHINE_ID_FILE = "/etc/machine-id" # use the dbus one as a fallback UNIQ_MACHINE_ID_FILE_FALLBACK = "/var/lib/dbus/machine-id" APP_INSTALL_PATTERN = "/usr/share/app-install/desktop/%s:*.desktop" # the configuration key to turn phased-updates always on ALWAYS_INCLUDE_PHASED_UPDATES = ( "Update-Manager::Always-Include-Phased-Updates") # ... or always off NEVER_INCLUDE_PHASED_UPDATES = ( "Update-Manager::Never-Include-Phased-Updates") def __init__(self, parent, dist=None): self.dist = (dist if dist else utils.get_dist()) self.distUpgradeWouldDelete = 0 self.oem_groups = [] self.update_groups = [] self.security_groups = [] self.kernel_autoremove_groups = [] self.duplicate_groups = [] self.ubuntu_pro_groups = [] self.ubuntu_pro_fake_groups = [] self.num_updates = 0 self.random = random.Random() self.ignored_phased_updates = [] # a stable machine uniq id try: with open(self.UNIQ_MACHINE_ID_FILE) as f: self.machine_uniq_id = f.read() except FileNotFoundError: with open(self.UNIQ_MACHINE_ID_FILE_FALLBACK) as f: self.machine_uniq_id = f.read() if 'XDG_DATA_DIRS' in os.environ and os.environ['XDG_DATA_DIRS']: data_dirs = os.environ['XDG_DATA_DIRS'] else: data_dirs = '/usr/local/share/:/usr/share/' self.application_dirs = [os.path.join(base, 'applications') for base in data_dirs.split(':')] if 'XDG_CURRENT_DESKTOP' in os.environ: self.current_desktop = os.environ.get('XDG_CURRENT_DESKTOP') else: self.current_desktop = '' self.desktop_cache = {} def _file_is_application(self, file_path): # WARNING: This is called often if there's a lot of updates. A poor # performing call here has a huge impact on the overall performance! if not file_path.endswith(".desktop"): # First the obvious case: If the path doesn't end in a .desktop # extension, this isn't a desktop file. return False file_path = os.path.abspath(file_path) for app_dir in self.application_dirs: if file_path.startswith(app_dir): return True return False def _rate_application_for_package(self, application, pkg): score = 0 desktop_file = os.path.basename(application.get_filename()) application_id = os.path.splitext(desktop_file)[0] if application.should_show(): score += 1 if application_id == pkg.name: score += 5 return score def _get_application_for_package(self, pkg): desktop_files = [] rated_applications = [] for installed_file in pkg.installed_files: if self._file_is_application(installed_file): desktop_files.append(installed_file) if pkg.name in self.desktop_cache: desktop_files += self.desktop_cache[pkg.name] for desktop_file in desktop_files: try: application = Gio.DesktopAppInfo.new_from_filename( desktop_file) application.set_desktop_env(self.current_desktop) except Exception as e: logging.warning("Error loading .desktop file %s: %s" % (desktop_file, e)) continue score = self._rate_application_for_package(application, pkg) if score > 0: rated_applications.append((score, application)) rated_applications.sort(key=lambda app: app[0], reverse=True) if len(rated_applications) > 0: return rated_applications[0][1] else: return None def _populate_desktop_cache(self, pkg_names): if not pkg_names: # No updates; This shouldn't have happened. logging.warning("_populate_desktop_cache called with empty list " "of packages.") return elif len(pkg_names) == 1: # One update; Let glob do the matching. pattern = self.APP_INSTALL_PATTERN % pkg_names[0] else: # More than one update available. Glob all desktop files and store # those that match an upgradeable package. pattern = self.APP_INSTALL_PATTERN % "*" for desktop_file in glob.iglob(pattern): try: pkg = desktop_file.split('/')[-1].split(":")[0] except IndexError: # app-install-data desktop file had an unexpected naming # convention. As we can't extract the package name from # the path, just ignore it. logging.error("Could not extract package name from '%s'. " "File ignored." % desktop_file) continue if pkg in pkg_names: self.desktop_cache.setdefault(pkg, []).append(desktop_file) logging.debug("App candidate for %s: %s" % (pkg, desktop_file)) def _is_security_update(self, pkg): """ This will test if the pkg is a security update. This includes if there is a newer version in -updates, but also an older update available in -security. For example, if installed pkg A v1.0 is available in both -updates (as v1.2) and -security (v1.1). we want to display it as a security update. :return: True if the update comes from the security pocket """ if not self.dist: return False inst_ver = pkg._pkg.current_ver for ver in pkg._pkg.version_list: # discard is < than installed ver if (inst_ver and apt.apt_pkg.version_compare(ver.ver_str, inst_ver.ver_str) <= 0): continue # check if we have a match for (verFileIter, index) in ver.file_list: if verFileIter.archive == "%s-security" % self.dist and \ verFileIter.origin == "Ubuntu": indexfile = pkg._pcache._list.find_index(verFileIter) if indexfile: # and indexfile.IsTrusted: return True return False def _is_ignored_phased_update(self, pkg): """ This will test if the pkg is a phased update and if it needs to get installed or ignored. :return: True if the updates should be ignored """ # allow the admin to override this if apt.apt_pkg.config.find_b( self.ALWAYS_INCLUDE_PHASED_UPDATES, False): return False if self.PHASED_UPDATES_KEY in pkg.candidate.record: if apt.apt_pkg.config.find_b( self.NEVER_INCLUDE_PHASED_UPDATES, False): logging.info("holding back phased update per configuration") return True # its important that we always get the same result on # multiple runs of the update-manager, so we need to # feed a seed that is a combination of the pkg/ver/machine self.random.seed("%s-%s-%s" % ( pkg.candidate.source_name, pkg.candidate.version, self.machine_uniq_id)) threshold = pkg.candidate.record[self.PHASED_UPDATES_KEY] percentage = self.random.randint(0, 100) if percentage > int(threshold): logging.info("holding back phased update %s (%s < %s)" % ( pkg.name, threshold, percentage)) return True return False def _make_groups(self, cache, pkgs, eventloop_callback, to_remove=False, sensitive=True): if not pkgs: return [] ungrouped_pkgs = [] app_groups = [] pkg_groups = [] for pkg in pkgs: app = self._get_application_for_package(pkg) if app is not None: app_group = UpdateApplicationGroup(pkg, app, to_remove, sensitive) app_groups.append(app_group) else: ungrouped_pkgs.append(pkg) # Stick together applications and their immediate dependencies for pkg in list(ungrouped_pkgs): dep_groups = [] for group in app_groups: if group.is_dependency(pkg, cache, eventloop_callback): dep_groups.append(group) if len(dep_groups) > 1: break if len(dep_groups) == 1: dep_groups[0].add(pkg, cache, eventloop_callback, to_remove) ungrouped_pkgs.remove(pkg) system_group = None if ungrouped_pkgs: # Separate out system base packages. If we have already found an # application for all updates, don't bother. linux_names = ("linux$", "linux-.*-buildinfo.*", "linux-.*-dev.*", "linux-.*-generic.*", "linux-.*-headers.*", "linux-.*-hwe.*", "linux-.*-gcp.*", "linux-.*-kernel.*", "linux-.*-lowlatency.*", "linux-.*-modules.*", "linux-.*-raspi.*", "linux-.*-tools.*", "linux-.*-virtual.*", "linux-base.*", "linux-crashdump.*", "linux-doc.*") linux_regexp = re.compile("(" + "|".join( ["^" + n for n in linux_names]) + ")") ubuntu_base_group = UpdateGroup(None, None, None, to_remove, sensitive) flavor_package = utils.get_ubuntu_flavor_package(cache=cache) ubuntu_base_pkgs = [flavor_package, "ubuntu-standard", "ubuntu-minimal"] for pkg in cache: if linux_regexp.match(pkg.name): ubuntu_base_pkgs.append(pkg.name) for pkg in ubuntu_base_pkgs: if pkg in cache and not to_remove: # the check for to_remove avoids creating groups in # kernel_autoremove and duplicate package sections # which fixes LP: #1912718 ubuntu_base_group.add(cache[pkg]) for pkg in ungrouped_pkgs: if ubuntu_base_group.is_dependency(pkg, cache, eventloop_callback): if system_group is None: system_group = UpdateSystemGroup(cache, to_remove, sensitive) system_group.add(pkg) else: pkg_groups.append(UpdatePackageGroup( pkg, to_remove, sensitive) ) app_groups.sort(key=lambda a: a.name.lower()) pkg_groups.sort(key=lambda a: a.name.lower()) if system_group: pkg_groups.append(system_group) return app_groups + pkg_groups def update(self, cache, eventloop_callback=None, duplicate_packages=[], ua_security_packages=[]): self.held_back = [] # do the upgrade self.distUpgradeWouldDelete = cache.saveDistUpgrade() oem_pkgs = [] pro_pkgs = [] security_pkgs = [] upgrade_pkgs = [] kernel_autoremove_pkgs = [] duplicate_pkgs = [] class FakeUbuntuProPackageCandidate: def __init__(self, source_name, version, size): self.source_name = source_name self.summary = source_name self.description = source_name self.version = version self.size = size self.downloadable = False self.record = {} class FakeUbuntuProPackage: def __init__(self, package_name, version, size): self.name = package_name self.candidate = FakeUbuntuProPackageCandidate(package_name, version, size) self.marked_install = False self.marked_upgrade = False self.marked_delete = False self.installed_files = [] def mark_install(self): pass def mark_delete(self): pass fake_ua_packages = [] ua_packages_names = [] for ( package_name, version, size, downloadable ) in ua_security_packages: if downloadable: ua_packages_names.append(package_name) else: fake_ua_packages.append(FakeUbuntuProPackage(package_name, version, size)) # Find all upgradable packages for pkg in cache: if pkg.is_upgradable or pkg.marked_install: if getattr(pkg.candidate, "origins", None) is None: # can happen for e.g. locked packages # FIXME: do something more sensible here (but what?) print("WARNING: upgradable but no candidate.origins?!?: ", pkg.name) continue # see if its a phased update and *not* a security update # keep track of packages for which the update percentage was # not met for testing is_security_update = self._is_security_update(pkg) if not is_security_update: if self._is_ignored_phased_update(pkg): self.ignored_phased_updates.append(pkg) continue if fnmatch.fnmatch(pkg.name, 'oem-*-meta'): oem_pkgs.append(pkg) comes_from_pro = pkg.name in ua_packages_names if comes_from_pro: pro_pkgs.append(pkg) elif is_security_update: security_pkgs.append(pkg) else: upgrade_pkgs.append(pkg) self.num_updates = self.num_updates + 1 if pkg.is_upgradable and not (pkg.marked_upgrade or pkg.marked_install): self.held_back.append(pkg.name) continue if (pkg.is_auto_removable and (cache.versioned_kernel_pkgs_regexp and cache.versioned_kernel_pkgs_regexp.match(pkg.name) and not cache.running_kernel_pkgs_regexp.match( pkg.name))): kernel_autoremove_pkgs.append(pkg) if (pkg.name in duplicate_packages): duplicate_pkgs.append(pkg) # perform operations after the loop to not skip packages which # changed state due to the resolver for pkg in kernel_autoremove_pkgs: pkg.mark_delete() for pkg in duplicate_pkgs: pkg.mark_delete() for pkg in self.ignored_phased_updates: pkg.mark_keep() if security_pkgs or upgrade_pkgs or pro_pkgs: # There's updates available. Initiate the desktop file cache. pkg_names = [p.name for p in pro_pkgs + security_pkgs + upgrade_pkgs + kernel_autoremove_pkgs + duplicate_pkgs] self._populate_desktop_cache(pkg_names) self.oem_groups = self._make_groups(cache, oem_pkgs, eventloop_callback) self.update_groups = self._make_groups(cache, upgrade_pkgs, eventloop_callback) self.security_groups = self._make_groups(cache, security_pkgs, eventloop_callback) self.kernel_autoremove_groups = self._make_groups( cache, kernel_autoremove_pkgs, eventloop_callback, True) self.duplicate_groups = self._make_groups( cache, duplicate_pkgs, eventloop_callback, True) self.ubuntu_pro_fake_groups = self._make_groups( cache, fake_ua_packages, eventloop_callback, sensitive=False) self.ubuntu_pro_groups = self._make_groups( cache, pro_pkgs, eventloop_callback)
| ver. 1.4 |
Github
|
.
| PHP 8.2.28 | Generation time: 0.03 |
proxy
|
phpinfo
|
Settings