File manager - Edit - /home/newsbmcs.com/public_html/static/img/logo/UpdateManager.tar
Back
UnitySupport.py 0000644 00000006506 15030301074 0007623 0 ustar 00 # UnitySupport.py # -*- Mode: Python; indent-tabs-mode: nil; tab-width: 4; coding: utf-8 -*- # # Copyright (c) 2011 Canonical # # Author: Michael Vogt <mvo@ubuntu.com> # Bilal Akhtar <bilalakhtar@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 import logging from gettext import gettext as _ HAVE_UNITY_SUPPORT = False try: import gi gi.require_version('Dbusmenu', '0.4') gi.require_version('Unity', '7.0') from gi.repository import Dbusmenu, Unity HAVE_UNITY_SUPPORT = True except (ValueError, ImportError) as e: logging.warning("can not import unity GI %s" % e) class IUnitySupport(object): """ interface for unity support """ def __init__(self, parent=None): pass def set_urgency(self, urgent): pass def set_install_menuitem_visible(self, visible): pass def set_progress(self, progress): pass class UnitySupportImpl(IUnitySupport): """ implementation of unity support (if unity is available) """ def __init__(self, parent=None): # create launcher and quicklist um_launcher_entry = Unity.LauncherEntry.get_for_desktop_id( "update-manager.desktop") self._unity = um_launcher_entry if parent: self._add_quicklist(parent) def _add_quicklist(self, parent): quicklist = Dbusmenu.Menuitem.new() # install self.install_dbusmenuitem = Dbusmenu.Menuitem.new() self.install_dbusmenuitem.property_set( Dbusmenu.MENUITEM_PROP_LABEL, _("Install All Available Updates")) self.install_dbusmenuitem.property_set_bool( Dbusmenu.MENUITEM_PROP_VISIBLE, True) self.install_dbusmenuitem.connect( "item-activated", parent.install_all_updates, None) quicklist.child_append(self.install_dbusmenuitem) # add it self._unity.set_property("quicklist", quicklist) def set_progress(self, progress): """ set the progress [0,100] """ self._unity.set_property("progress", progress / 100.0) # hide progress when out of bounds if progress < 0 or progress > 100: self._unity.set_property("progress_visible", False) else: self._unity.set_property("progress_visible", True) def set_urgency(self, urgent): self._unity.set_property("urgent", urgent) def set_install_menuitem_visible(self, visible): self.install_dbusmenuitem.property_set_bool( Dbusmenu.MENUITEM_PROP_VISIBLE, visible) # check what to export to the clients if HAVE_UNITY_SUPPORT: UnitySupport = UnitySupportImpl else: # we just provide the empty interface UnitySupport = IUnitySupport UpdateManagerVersion.py 0000644 00000000027 15030301074 0011171 0 ustar 00 VERSION = '1:22.04.21' __pycache__/__init__.cpython-310.pyc 0000644 00000000225 15030301074 0013204 0 ustar 00 o ��{f � @ s d S )N� r r r �8/usr/lib/python3/dist-packages/UpdateManager/__init__.py�<module> s __pycache__/UpdateManagerVersion.cpython-310.pyc 0000644 00000000267 15030301074 0015536 0 ustar 00 o ��{f � @ s d Z dS )z 1:22.04.21N)�VERSION� r r �D/usr/lib/python3/dist-packages/UpdateManager/UpdateManagerVersion.py�<module> s __pycache__/UnitySupport.cpython-310.pyc 0000644 00000005611 15030301074 0014156 0 ustar 00 o ��{fF � @ s� d dl Z d dlmZ dZz d dlZe�dd� e�dd� d dlmZmZ d ZW n e e fyC Z z e � d e � W Y dZ [ ndZ [ ww G d d � d e �ZG d d� de�ZerZeZdS eZdS )� N)�gettextF�Dbusmenuz0.4�Unityz7.0)r r Tzcan not import unity GI %sc @ s2 e Zd ZdZd dd�Zdd� Zdd� Zd d � ZdS ) � IUnitySupportz interface for unity support Nc C � d S �N� )�self�parentr r �</usr/lib/python3/dist-packages/UpdateManager/UnitySupport.py�__init__( � zIUnitySupport.__init__c C r r r �r �urgentr r r � set_urgency+ r zIUnitySupport.set_urgencyc C r r r �r �visibler r r � set_install_menuitem_visible. r z*IUnitySupport.set_install_menuitem_visiblec C r r r �r �progressr r r � set_progress1 r zIUnitySupport.set_progressr )�__name__� __module__� __qualname__�__doc__r r r r r r r r r & s r c @ s: e Zd ZdZd dd�Zdd� Zdd� Zd d � Zd d � ZdS )�UnitySupportImplz9 implementation of unity support (if unity is available) Nc C s( t j�d�}|| _|r| �|� d S d S )Nzupdate-manager.desktop)r � LauncherEntry�get_for_desktop_id�_unity�_add_quicklist)r r �um_launcher_entryr r r r 8 s ��zUnitySupportImpl.__init__c C sj t j�� }t j�� | _| j�t jtd�� | j�t jd� | j� d|j d � |� | j� | j � d|� d S )Nz Install All Available UpdatesTzitem-activated� quicklist)r �Menuitem�new�install_dbusmenuitem� property_set�MENUITEM_PROP_LABEL�_�property_set_bool�MENUITEM_PROP_VISIBLE�connect�install_all_updates� child_appendr � set_property)r r r! r r r r @ s ��� z UnitySupportImpl._add_quicklistc C sF | j �d|d � |dk s|dkr| j �dd� dS | j �dd� dS ) z set the progress [0,100] r g Y@r �d �progress_visibleFTN�r r- r r r r r O s z UnitySupportImpl.set_progressc C s | j �d|� d S )Nr r0 r r r r r X s z UnitySupportImpl.set_urgencyc C s | j �tj|� d S r )r$ r( r r) r r r r r [ s �z-UnitySupportImpl.set_install_menuitem_visibler ) r r r r r r r r r r r r r r 5 s r )�loggingr r' �HAVE_UNITY_SUPPORT�gi�require_version� gi.repositoryr r � ValueError� ImportError�e�warning�objectr r � UnitySupportr r r r �<module> s$ ��, Core/__pycache__/__init__.cpython-310.pyc 0000644 00000000232 15030301074 0014072 0 ustar 00 o ��{f � @ s d S )N� r r r �=/usr/lib/python3/dist-packages/UpdateManager/Core/__init__.py�<module> s Core/__pycache__/roam.cpython-310.pyc 0000644 00000011044 15030301074 0013274 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 � Core/__pycache__/MetaRelease.cpython-310.pyc 0000644 00000022407 15030301074 0014532 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� Core/__pycache__/UpdateList.cpython-310.pyc 0000644 00000040160 15030301074 0014415 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 Core/__pycache__/utils.cpython-310.pyc 0000644 00000035536 15030301074 0013512 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 � +, � Core/__pycache__/MyCache.cpython-310.pyc 0000644 00000030127 15030301074 0013652 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 � � � Core/__pycache__/AlertWatcher.cpython-310.pyc 0000644 00000005436 15030301074 0014733 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 Core/__pycache__/LivePatchSocket.cpython-310.pyc 0000644 00000010277 15030301074 0015375 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 : Core/MetaRelease.py 0000644 00000042207 15030301074 0010173 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) Core/MyCache.py 0000644 00000044705 15030301074 0007322 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 Core/roam.py 0000644 00000017772 15030301074 0006753 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) Core/LivePatchSocket.py 0000644 00000007322 15030301074 0011033 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 Core/AlertWatcher.py 0000644 00000007765 15030301074 0010403 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) Core/__init__.py 0000644 00000000000 15030301074 0007524 0 ustar 00 Core/utils.py 0000644 00000044155 15030301074 0007150 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") Core/UpdateList.py 0000644 00000060126 15030301074 0010062 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) backend/__pycache__/__init__.cpython-310.pyc 0000644 00000022235 15030301074 0014600 0 ustar 00 o ��{f�@ � @ s\ d Z ddlmZ ddlZe�dd� e�dd� ddlmZmZmZ dd l m Z ddl Z ddl Z ddl Z ddlZddlZdd lmZ dd lmZ dd lmZ dd lmZmZ ddlmZ G dd� de�Ze j � d�r�de j vr�zddl m Z W n e y� e �!d� Y nw e j � d�r�de j vr�zddl"m"Z" W n e y� e �!d� Y nw dd� Z#dS ) z2Integration of package managers into UpdateManager� )�absolute_importN�Gtkz3.0�Snapd�1)�GLibr r )�Cache)�gettext)�Thread)�MyCache)� inhibit_sleep�get_dist_version)�Dialogc @ st e Zd ZdZdZdZdd� Zdd� Zdd � Zd d � Z d d � Z dd� Z dd� Z dd� Z dd� Zdd� Z d dd�ZdS ) �InstallBackendr � � c C s t �| |� || _d | _d S �N)r �__init__�action� sleep_cookie)�self� window_mainr � r �@/usr/lib/python3/dist-packages/UpdateManager/backend/__init__.pyr ! s zInstallBackend.__init__c C st dt jd< t� | _| j| jkrM| jj}t| jjj d�}g }g }| jj D ]!}|| }|| }|j r9|j s9|� |� q#|jrD|jrD|� |� q#| �||� d S | j| jkr�g }g } g } t| jjj d�}| jjD ]E}z:|j r�||j j s�|j} |jr{| d7 } |� | � n |jr�||j jr�| � |j� n|jr�||j j r�| � |j� W qe ty� Y qew | �|| | � d S | �� d S )N�none�APT_LISTCHANGES_FRONTEND)�rootdirz#auto)�os�environr r r �ACTION_PRE_INSTALLr �cacher r �oem_metapackages�marked_install� is_installed�append�marked_upgrade� is_upgradable� commit_oem�ACTION_INSTALL�name�is_auto_installed� marked_delete�KeyError�commit�update) r � unfresh_cache� fresh_cache�pkgs_install_oem�pkgs_upgrade_oem�pkg� unfresh_pkg� fresh_pkg� pkgs_install� pkgs_upgrade� pkgs_remove�pkgnamer r r �start& s` � � � � � � � � zInstallBackend.startc C � t �)z(Run a update to refresh the package list��NotImplementedError�r r r r r- Z � zInstallBackend.updatec C s | j | jdddddd� dS )z Install these OEM packages TN)� authorized�success� error_string� error_desc� trans_failed)� _action_doner )r r0 r1 r r r r& ^ s �zInstallBackend.commit_oemc C r: )zCommit the cache changes r; )r r5 r6 r7 r r r r, e r> zInstallBackend.commitc C s: i }i }dt � }| jj}zstdd�}t�|�}|�� |d D ]/}|d | }|�dd �} |�d|�} |�dd �} | |vr?q | rI|| jdu rIq | | f||< q |d D ]'}|d | } | �d |�} |�dd �} | |vrmqT| rw|| jdu rwqT| ||< qTW ||fS t y� } zt � d | � W Y d }~||fS d }~ww ) Nzstable/ubuntu-z0/usr/share/ubuntu-release-upgrader/deb2snap.json�r�seeded�deb� to_channel�metapkgF�unseeded� from_channelz%error reading deb2snap.json file (%s)) r r r �open�json�load�close�getr" � Exception�logging�debug)r � seeded_snaps�unseeded_snaps� curr_channelr �d2s_file�d2s�snap�seedrG rH rI �unseedrK �er r r �get_snap_seedsi sH � � ���z InstallBackend.get_snap_seedsc C s z | j jd u rtd �| j _n | j j�d � | j j�� | j j}W n ty; } zt�d| � g W Y d }~S d }~ww g }| �� \}}|� � D ]6\}\}}||v r~|| j r~d}|| � � |� � D ]} | j rr| j rr| jsrd} nqc|�� |r~|�|� qH|S )Nzerror reading cache (%s)TF)r r r rL � _initDepCacherQ rR rS r] �itemsr" � mark_delete� get_changesr* r) �clearr# ) r r r\ � duplicatesrT �_rY rG � deb_is_autor2 r r r �get_deb2snap_dups� s: �� �� �z InstallBackend.get_deb2snap_dupsc C s� i }| � � \}}|�� D ]:\}\}}i }tjdd|gdtjd��� }t�d|d tj�r4t � d| � q || j j v rFd|d < ||d < |||< q |�� D ]�\}} i }tjdd|gdtjd��� }t�d|d tj�r�t � d| � d } t�d | |d tj�s�t � d | � qKd|d < tjdd|gdtjd��� } | d � d�D ]>} | � � } t| �dkr�q�| d }| d }|�|d �r�|� d�d }|dkr�||vr�t � d||f � d|d < d|d < nq�|||< qK|S )NrY �infoT)�universal_newlines�stdoutz ^installed: r zSnap %s is installed�install�command�channelz#stable/ubuntu-[0-9][0-9].[0-9][0-9]z^tracking:.*%sz+Snap %s is not tracking the release channel�remove� connections� � r r �:�-z9Snap %s is being used by %s. Switching it to stable track�refresh�stable)r] r_ � subprocess�Popen�PIPE� communicate�re�search� MULTILINErR rS r �duplicate_packages�split�len� startswith)r � snap_listrT rU rY rG rH � snap_object� snap_inforK � re_channel�conns�conn� conn_cols�plug�slot� plug_snapr r r �get_snap_transitions� s~ �� � �� �� �� ���z#InstallBackend.get_snap_transitionsc C sz |\}}}|s d S d}d} |� � D ]} || �� 7 }| | �� 7 } q| | } | | || } | |�� kr;t�|j| � d S d S )Nr )� get_tasks�get_progress_total�get_progress_done� get_fractionr �idle_add� set_fraction) r �client�changerd � user_data�index�count� progress_bar� task_total� task_done�task� task_fraction�total_fractionr r r �update_snap_cb� s �z InstallBackend.update_snap_cbc s� � fdd�}dd� }|t d�� d }d }� j�� }|r.t|d tj�r.|d }t�d||�}� �� }|r>t� |� |� d� zmt � � }|� � d}t|�} |�� D ]X\} } | d } | d kru|t d | �� |j| | d � j|| |fd � n0| d kr�|t d| �� |j| � j|| |fd � n|t d| �� |j| | d � j|| |fd � |d7 }qQW n) tjy� } z t�d| � t�� jjdt d�t d�� W Y d } ~ d S d } ~ ww t�� jj� d S )Nc s t �� jj| � d S r )r r� � label_details� set_label)�statusr= r r � update_status s z2InstallBackend.update_snaps.<locals>.update_statusc S s | � � dS )NT)�pulse)r� r r r �update_progress s z4InstallBackend.update_snaps.<locals>.update_progresszUpdating snapsr �d rk rs zRefreshing %s snaprl )�progress_callback_datarm zRemoving %s snapzInstalling %s snapr zerror updating snaps (%s)Fz!Upgrade only partially completed.zMAn error occurred while updating snaps. Please check your network connection.)rd �progressbar_slot� get_children� isinstancer � ProgressBarr � timeout_addr� � source_remover� r �Client� connect_syncr~ r_ � refresh_syncr� � remove_sync� install_sync�ErrorrR rS r� r � start_error�start_available)r r� r� r� �progress_timer� progress_barsr� r� r� r� rY r� rk r\ r r= r � update_snaps sp � �� �� �� � ��� zInstallBackend.update_snapsFc C s� || j kr|r| j| _| �� d S || jkrL|r-tj�d�r-t| d�r-t| j d��� d S |r6| j � � d S |rB| j � |||� d S t �d| j j� d S |rX| j � d||� d S |rrtj�d�rrt| d�rr| �� | j _| j � � d S | }| j � |� d S )Nz /usr/bin/snap�pane_update_progress)�targetr T)r r' r r9 r �path�exists�hasattrr r� r r� r� r r� �exitrf r| )r r r? r@ rA rB rC �is_cancelled_updater r r rD L s0 � �� zInstallBackend._action_doneN)F)�__name__� __module__� __qualname__� ACTION_UPDATEr r' r r9 r- r& r, r] rf r� r� r� rD r r r r r s 4&&AA�r �/usr/sbin/aptd�%UPDATE_MANAGER_FORCE_BACKEND_SYNAPTICr )�InstallBackendAptdaemonzimporting aptdaemon�/usr/sbin/synaptic�&UPDATE_MANAGER_FORCE_BACKEND_APTDAEMON)�InstallBackendSynapticzimporting synapticc O s� t j�d�r"dt jvr"zt| i |��W S ty! t�d� Y nw t j�d�rBdt jvrBzt| i |��W S tyA Y t d��w t d��)z,Select and return a package manager backend.r� r� zusing aptdaemon failedr� r� zENo working backend found, please try installing aptdaemon or synaptic) r r� r� r r� � NameErrorrR � exceptionr� rQ )�args�kwargsr r r � get_backend� s � �r� )$�__doc__� __future__r �gi�require_version� gi.repositoryr r r �aptr rM rR r ry ru r rd � threadingr �UpdateManager.Core.MyCacher �UpdateManager.Core.utilsr r �UpdateManager.Dialogsr r r� r� r r� � ImportErrorr� r� r� r r r r �<module> sF Z � � backend/__pycache__/InstallBackendSynaptic.cpython-310.pyc 0000644 00000004761 15030301074 0017436 0 ustar 00 o ��{f� � @ sl d dl Z d dlZd dlmZ d dlZe�dd� d dlmZ d dlmZ e d dl m Z G dd � d e �Z dS ) � N)�gettext�Gtkz3.0)�GObject)�GdkX11)�InstallBackendc @ s4 e Zd ZdZdd� Z d dd�Zdd� Zd d � Zd S ) �InstallBackendSynapticz# Install backend based on synaptic c C s dg}d }| � | j||� d S )Nz--update-at-startup)� _run_synaptic� ACTION_UPDATE)�self�opt�tempf� r �N/usr/lib/python3/dist-packages/UpdateManager/backend/InstallBackendSynaptic.py�update s z InstallBackendSynaptic.updateFc C s� g }|r|� d� |� d� |� d� |� dtd� � |� d� |� dtd� � tjdd �}|| D ] }|�d | � q4|D ] }|�d | � q@|� d � |� d|j � |�� | �| j||� d S ) Nz-ozSynaptic::closeZvt=truez--progress-str�%sz%Please wait, this can take some time.z --finish-strzUpdate is completezw+)�modez %s install z %s deinstall z--set-selections-file) �append�_�tempfile�NamedTemporaryFile�write�name�flushr �ACTION_INSTALL)r � pkgs_install� pkgs_upgrade� pkgs_remove� close_on_doner r �pkg_namer r r �commit s" z InstallBackendSynaptic.commitc C s� | j �� }z|�� }W n ty d}Y nw dddddd| g}|�|� tj}tj||d�\}} } } ||f| _t� || j | j� d S ) zExecute synaptic.r z/usr/bin/pkexecz/usr/sbin/synapticz--hide-main-windowz--non-interactivez--parent-window-idr )�flagsN) � window_main� get_window�get_xid�AttributeError�extendr �SPAWN_DO_NOT_REAP_CHILD� spawn_async� child_data�child_watch_add�_on_synaptic_exit) r �actionr r �win�xid�cmdr �pid�stdin�stdout�stderrr r r r 4 s �� z$InstallBackendSynaptic._run_synapticc C s6 |\}}|r |� � | j|dt�|�dkd d d� d S )NTr )� authorized�success� error_string� error_desc)�close� _action_done�os� WEXITSTATUS)r r/ � condition�datar+ r r r r r* G s �z(InstallBackendSynaptic._on_synaptic_exitN)F)�__name__� __module__� __qualname__�__doc__r r r r* r r r r r s � r ) r9 r r r �gi�require_version� gi.repositoryr r �UpdateManager.backendr r r r r r �<module> s backend/__pycache__/InstallBackendAptdaemon.cpython-310.pyc 0000644 00000031734 15030301074 0017554 0 ustar 00 o ��{f�L � @ sL d dl mZ d dlmZ d dlmZmZ d dlmZ d dl m Z m Z m Z m Z mZ d dlmZmZmZmZmZmZmZmZmZ d dlmZ d dlm Z d d l m Z d d l m Z d d l!Z!d d l"Z"G d d � d e �Z#G dd� dej$�Z%G dd� dee �Z&e'dkr�d d l(Z(e(�)� Z*dZ+d dl,m-Z- e-e+e*�Z.e&e.d �Z/e/�0dgg g � e�1� d S d S )� )�print_function)�Gtk)�client�errors)�inline_callbacks)�AptCancelButton�AptConfigFileConflictDialog�AptDetailsExpander�AptMediumRequiredDialog�AptProgressBar) � EXIT_SUCCESS� EXIT_FAILED�STATUS_COMMITTING�STATUS_DOWNLOADING�STATUS_DOWNLOADING_REPO�STATUS_FINISHED� get_error_description_from_enum�get_error_string_from_enum�get_status_string_from_enum)�InstallBackend)� UnitySupport)� BuilderDialog)�gettextNc s* e Zd ZdZd� fdd� Zdd� Z� ZS ) �UpdateManagerExpanderaF An AptDetailsExpander which can be used with multiple terminals. The default AptDetailsExpander will shrink/hide when its transaction finishes. But here we want to support "chaining" transactions. So we override the status-changed handler to only do that when we are running the final transaction.TFc s t � �||� || _d S �N)�super�__init__�final)�self� transaction�terminalr �� __class__� �O/usr/lib/python3/dist-packages/UpdateManager/backend/InstallBackendAptdaemon.pyr - s z UpdateManagerExpander.__init__c C s� |t tfv r | �d� | j�� | jr| j�� d S d S |tkrE| j�� | jr4| j�� | �d� d S | jrC| � d� | �d� d S d S | jre|t krg| j�� | jrY| j�� | �d� | � d� d S d S d S �NTF) r r � set_sensitive�download_scrolled�showr �hider r � set_expandedr )r �trans�statusr# r# r$ �_on_status_changed1 s, � � �z(UpdateManagerExpander._on_status_changed)TF)�__name__� __module__� __qualname__�__doc__r r- � __classcell__r# r# r! r$ r % s r c s8 e Zd ZdZ� fdd�Zdd� Zdd� Zdd � Z� ZS ) �AptStackedProgressBaraq A GtkProgressBar which represents the state of many aptdaemon transactions. aptdaemon provides AptProgressBar for the state of *one* transaction to be represented in a progress bar. This widget creates one of those per containing transaction, and scales its progress to the given ratio, so one progress bar can show the state of many transactions.c s d| _ g | _|| _t� �� d S )Nr )�current_max_progress� progress_bars�unityr r )r r6 r! r# r$ r P s z AptStackedProgressBar.__init__c C s~ d| kr dksJ � J �t |�}| j�|� | j|_| j|7 _| jdkr+d| _| j|_|�d| j� |�d| j� d S )Nr � znotify::fractionz notify::text) r r5 �appendr4 �min�max�connect�_update_progress� _update_text)r r+ � max_progress�progressr# r# r$ �add_transactionW s z%AptStackedProgressBar.add_transactionc C s@ |j |j }||�� }|j| }| �|� | j�|d � d S )N�d )r: r9 � get_fraction� set_fractionr6 � set_progress)r �inner_progress�data�delta�position_in_delta� new_progressr# r# r$ r< f s z&AptStackedProgressBar._update_progressc C s | � |�� � d S r )�set_text�get_text)r rE rF r# r# r$ r= m s z"AptStackedProgressBar._update_text) r. r/ r0 r1 r r@ r< r= r2 r# r# r! r$ r3 G s r3 c @ s� e Zd ZdZdd� Zdd� Zedd� �Zdd � Zd d � Z ed d � �Z dd� Z edd� �Z edd� �Z dd� Zdd� Ze d&dd��Zd d � Zd d � Zd d!� Zd"d#� Zd$d%� ZdS )'�InstallBackendAptdaemonzCMakes use of aptdaemon to refresh the cache and to install updates.c C sh t �| ||� tj�|jd�}t�| ||d� t�� | _t � | _ d | _ d | _ d | _ d | _d | _d | _d S )Nz gtkbuilder/UpdateProgress.ui�pane_update_progress)r r �os�path�join�datadirr r � AptClientr r6 �_expanded_size� button_cancel�trans_failed_msg� progressbar�_active_transaction� _expander)r � window_main�action�ui_pathr# r# r$ r t s � � z InstallBackendAptdaemon.__init__c C s: | j r| j �� rz| j �� W dS ty Y dS w dS r% )rT � get_sensitive�clicked� Exception)r r# r# r$ �close� s ��z InstallBackendAptdaemon.closec c s� � z| j jdd�V }| �|| jtd�d�V W dS tjy. | j| jddddd� Y dS ty@ | j| jddddd� � w )zRefresh the package listT��deferu Checking for updates…FN�� authorized�success� error_string� error_desc) r � update_cache�_show_transaction� ACTION_UPDATE�_r �NotAuthorizedError� _action_doner^ )r r+ r# r# r$ �update� s$ � � � ��z InstallBackendAptdaemon.updatec C sP t |jj�}t|jj�}| jrd}|d | j }nd }| j|dd|||d� d S )NT� F�rc rd re rf � trans_failed)r �error�coder rU rl )r r+ rZ re rf rp r# r# r$ �_show_transaction_error� s �z/InstallBackendAptdaemon._show_transaction_errorc C s$ |t kr | �||� d S | �� d S r )r rs �_apt_update_oem)r r+ r, rZ r# r# r$ �_update_next_package� s z,InstallBackendAptdaemon._update_next_packagec c sN � | j sJ �| j �� }d|� d�}z0tj�|�r?| jj|d�V }| j r'| j}n| j}| j || j t d�d|d| j d�V W d S W d S t jyW | j| j ddd d d � Y d S t jyo } z t|�| _W Y d }~d S d }~w tjy� } z|�� d kr~� | j| j ddd d d � W Y d }~d S d }~w ty� | j| j ddd d d � � w ) Nz/etc/apt/sources.list.d/z.list)� sources_list� Installing updates…T皙�����?��on_finished_handler�progress_bar_maxFrb �"org.freedesktop.DBus.Error.NoReply)�_oem_packages_to_update�poprN rO �existsr rg ru � _on_finishedrh �ACTION_PRE_INSTALLrj �_len_oem_updatesr rk rl �TransactionFailed�strrU �dbus� DBusException� get_dbus_namer^ )r �elem�sources_list_filer+ �finished_handler�er# r# r$ rt � sV � ��� �� �� ��z'InstallBackendAptdaemon._apt_update_oemc C sJ |t kr | �||� d S |j\}}}}}}t|�| _t|�| _| �� d S r )r rs �packages�setr} �lenr� rt )r r+ r, rZ �installrj r# r# r$ � _update_oem� s z#InstallBackendAptdaemon._update_oemc c sH � t |�t |�B | _z4|s |s | j| jddd d d d� W d S |r<| jj|dd�V }| j|| jtd�d| jdd�V W d S W d S t j yT | j| jddd d d� Y d S t j yl } z t |�| _ W Y d }~d S d }~w tjy� } z|�� d kr{� | j| jddd d d� W Y d }~d S d }~w ty� | j| jddd d d� � w ) NTro r` rw rx ry Frb r| )r� �all_oem_packagesrl r� r �install_packagesrh rj r� r rk r� r� rU r� r� r� r^ )r �pkgs_install_oem�pkgs_upgrade_oemr+ r� r# r# r$ � commit_oem� sZ ����� �� �� ��z"InstallBackendAptdaemon.commit_oemc c s � z#g } }}| j j||||||dd�V }| �|| jtd�d�V W dS tjy: | j| jddddd� Y dS tjyR } z t |�| _ W Y d}~dS d}~w t j yw } z|� � dkra� | j| jddddd� W Y d}~dS d}~w ty� | j| jddddd� � w )z)Commit a list of package adds and removesTr` rw FNrb r| )r �commit_packagesrh �ACTION_INSTALLrj r rk rl r� r� rU r� r� r� r^ ) r � pkgs_install� pkgs_upgrade� pkgs_remove� reinstall�purge� downgrader+ r� r# r# r$ �commit sB � � � �� �� ��z InstallBackendAptdaemon.commitc C s |� |� d S r )� set_label)r r+ �details� label_detailsr# r# r$ �_on_details_changed3 s z+InstallBackendAptdaemon._on_details_changedc C sD |� t|�� |tkr | jr | jj�� r | �| j� d S d S d S d S r )r� r r rX r � get_visible�_resize_to_show_details)r r+ r, r� r# r# r$ r- 6 s � �z*InstallBackendAptdaemon._on_status_changedr7 Nc c s� � |d u r| j }| j�|� | js#t| j�| _| j�� | j�| j� | j� ||� | j r4| j � |� nt |�| _ | j �� | j �| j � || jkrQ| j �tj� |r�| js�t|�| _| j�d� | j�d� | j�� | j�d| j� | j�| j� | j�� n| j� |� || jk| j_n| jr�| j� � |�d| j | j � |�d| j | j � |�d||� |�d| j � |�d| j!� |�"d�V |�#� V d S ) NTznotify::expandedzstatus-details-changedzstatus-changed�finishedzmedium-requiredzconfig-file-conflict�gnome)$r� � label_headerr� rV r3 r6 r( �progressbar_slot�addr@ rT �set_transactionr �button_cancel_slotri r � STOCK_STOPrX r � set_vexpand� set_hexpand�show_allr; � _on_expanded� expander_slotr� r �_expander_slotr) r� r� r- �_on_medium_required�_on_config_file_conflict�set_debconf_frontend�run)r r+ rZ �header� show_detailsr{ rz r# r# r$ rh > sT � � � � z)InstallBackendAptdaemon._show_transactionc C s| |� � s|j�� | j�� f| _| j�� d S | jr7| j\}\}}||j�� kr.| �|� d S | j�||� d S | �|� d S r ) � get_expandedr r� rY �get_sizerS �end_user_resizabler� �begin_user_resizable)r �expander�param� term_visible� stored_width� stored_heightr# r# r$ r� t s ��z$InstallBackendAptdaemon._on_expandedc C s� |� � rF| j�� \}}|�� j}|�� j}|j�� r5|j�� d }|j� � d }|| | }|| | } n|d }|d } | j� || � dS dS )a� Resize the window to show the expanded details. Unfortunately the expander only expands to the preferred size of the child widget (e.g showing all 80x24 chars of the Vte terminal) if the window is rendered the first time and the terminal is also visible. If the expander is expanded afterwards the window won't change its size anymore. So we have to do this manually. See LP#840942 �P � rA �� N) r� rY r� �get_allocation�width�heightr r� �get_char_width�get_char_heightr� ) r r� � win_width� win_height� exp_width� exp_height�terminal_width�terminal_height� new_width� new_heightr# r# r$ r� � s �z/InstallBackendAptdaemon._resize_to_show_detailsc C sD t ||| j�}|�� }|�� |tjjkr |�|� d S |�� d S r ) r rY r� r) r � ResponseType�OK�provide_medium�cancel)r r �medium�drive�dialog�resr# r# r$ r� � s z+InstallBackendAptdaemon._on_medium_requiredc C sJ t ||| j�}|�� }|�� |tjjkr |�|d� d S |�|d� d S )N�replace�keep)r rY r� r) r r� �YES� resolve_config_file_conflict)r r �old�newr� r� r# r# r$ r� � s z0InstallBackendAptdaemon._on_config_file_conflictc C s� d }d }d}|t kr"t|jj�}t|jj�}| jr"d}|d | j }| j�d� |tk}z| j |d||||d� W d S t yN | j |d|||d� Y d S w )NFTrn ���ro rb ) r r rq rr r rU r6 rD r rl � TypeError)r r+ r, rZ re rf rp � is_successr# r# r$ r� � s. � ��z$InstallBackendAptdaemon._on_finished)r7 N)r. r/ r0 r1 r r_ r rm rs ru rt r� r� r� r� r- rh r� r� r� r� r� r# r# r# r$ rL q s4 ( ' �5 rL �__main__z/usr/share/update-manager)� UpdateManager�2vcard)2� __future__r � gi.repositoryr � aptdaemonr r ra r �aptdaemon.gtk3widgetsr r r r r �aptdaemon.enumsr r r r r r r r r �UpdateManager.backendr �UpdateManager.UnitySupportr �UpdateManager.Dialogsr r rj r� rN r � ProgressBarr3 rL r. �mock�Mock�options�data_dir�UpdateManager.UpdateManagerr� �app�br� �mainr# r# r# r$ �<module> s6 , "* ^ � backend/InstallBackendAptdaemon.py 0000644 00000046245 15030301074 0013220 0 ustar 00 #!/usr/bin/env python # -*- Mode: Python; indent-tabs-mode: nil; tab-width: 4; coding: utf-8 -*- # (c) 2005-2012 Canonical, GPL # (C) 2008-2009 Sebastian Heinlein <devel@glatzor.de> from __future__ import print_function from gi.repository import Gtk from aptdaemon import client, errors from defer import inline_callbacks from aptdaemon.gtk3widgets import (AptCancelButton, AptConfigFileConflictDialog, AptDetailsExpander, AptMediumRequiredDialog, AptProgressBar) from aptdaemon.enums import (EXIT_SUCCESS, EXIT_FAILED, STATUS_COMMITTING, STATUS_DOWNLOADING, STATUS_DOWNLOADING_REPO, STATUS_FINISHED, get_error_description_from_enum, get_error_string_from_enum, get_status_string_from_enum) from UpdateManager.backend import InstallBackend from UpdateManager.UnitySupport import UnitySupport from UpdateManager.Dialogs import BuilderDialog from gettext import gettext as _ import dbus import os class UpdateManagerExpander(AptDetailsExpander): """An AptDetailsExpander which can be used with multiple terminals. The default AptDetailsExpander will shrink/hide when its transaction finishes. But here we want to support "chaining" transactions. So we override the status-changed handler to only do that when we are running the final transaction.""" def __init__(self, transaction, terminal=True, final=False): super().__init__(transaction, terminal) self.final = final def _on_status_changed(self, trans, status): if status in (STATUS_DOWNLOADING, STATUS_DOWNLOADING_REPO): self.set_sensitive(True) self.download_scrolled.show() if self.terminal: self.terminal.hide() elif status == STATUS_COMMITTING: self.download_scrolled.hide() if self.terminal: self.terminal.show() self.set_sensitive(True) elif self.final: self.set_expanded(False) self.set_sensitive(False) elif self.final and status == STATUS_FINISHED: self.download_scrolled.hide() if self.terminal: self.terminal.hide() self.set_sensitive(False) self.set_expanded(False) class AptStackedProgressBar(Gtk.ProgressBar): """ A GtkProgressBar which represents the state of many aptdaemon transactions. aptdaemon provides AptProgressBar for the state of *one* transaction to be represented in a progress bar. This widget creates one of those per containing transaction, and scales its progress to the given ratio, so one progress bar can show the state of many transactions.""" def __init__(self, unity): self.current_max_progress = 0 self.progress_bars = [] self.unity = unity super().__init__() def add_transaction(self, trans, max_progress): assert 0 <= max_progress <= 1 progress = AptProgressBar(trans) self.progress_bars.append(progress) progress.min = self.current_max_progress self.current_max_progress += max_progress if self.current_max_progress > 1: self.current_max_progress = 1 progress.max = self.current_max_progress progress.connect("notify::fraction", self._update_progress) progress.connect("notify::text", self._update_text) def _update_progress(self, inner_progress, data): delta = inner_progress.max - inner_progress.min position_in_delta = delta * inner_progress.get_fraction() new_progress = inner_progress.min + position_in_delta self.set_fraction(new_progress) self.unity.set_progress(new_progress * 100) def _update_text(self, inner_progress, data): self.set_text(inner_progress.get_text()) class InstallBackendAptdaemon(InstallBackend, BuilderDialog): """Makes use of aptdaemon to refresh the cache and to install updates.""" def __init__(self, window_main, action): InstallBackend.__init__(self, window_main, action) ui_path = os.path.join(window_main.datadir, "gtkbuilder/UpdateProgress.ui") BuilderDialog.__init__(self, window_main, ui_path, "pane_update_progress") self.client = client.AptClient() self.unity = UnitySupport() self._expanded_size = None self.button_cancel = None self.trans_failed_msg = None self.progressbar = None self._active_transaction = None self._expander = None def close(self): if self.button_cancel and self.button_cancel.get_sensitive(): try: self.button_cancel.clicked() except Exception: # there is not much left to do if the transaction can't be # canceled pass return True else: return False @inline_callbacks def update(self): """Refresh the package list""" try: trans = yield self.client.update_cache(defer=True) yield self._show_transaction(trans, self.ACTION_UPDATE, _("Checking for updates…"), False) except errors.NotAuthorizedError: self._action_done(self.ACTION_UPDATE, authorized=False, success=False, error_string=None, error_desc=None) except Exception: self._action_done(self.ACTION_UPDATE, authorized=True, success=False, error_string=None, error_desc=None) raise def _show_transaction_error(self, trans, action): error_string = get_error_string_from_enum(trans.error.code) error_desc = get_error_description_from_enum(trans.error.code) if self.trans_failed_msg: trans_failed = True error_desc = error_desc + "\n" + self.trans_failed_msg else: trans_failed = None self._action_done(action, authorized=True, success=False, error_string=error_string, error_desc=error_desc, trans_failed=trans_failed) def _update_next_package(self, trans, status, action): if status == EXIT_FAILED: self._show_transaction_error(trans, action) return self._apt_update_oem() @inline_callbacks def _apt_update_oem(self): assert self._oem_packages_to_update elem = self._oem_packages_to_update.pop() sources_list_file = f'/etc/apt/sources.list.d/{elem}.list' try: if os.path.exists(sources_list_file): trans = yield self.client.update_cache( sources_list=sources_list_file ) if self._oem_packages_to_update: finished_handler = self._update_next_package else: finished_handler = self._on_finished yield self._show_transaction( trans, self.ACTION_PRE_INSTALL, _("Installing updates…"), True, on_finished_handler=finished_handler, progress_bar_max=0.1 / self._len_oem_updates ) except errors.NotAuthorizedError: self._action_done(self.ACTION_PRE_INSTALL, authorized=False, success=False, error_string=None, error_desc=None) except errors.TransactionFailed as e: self.trans_failed_msg = str(e) except dbus.DBusException as e: if e.get_dbus_name() != "org.freedesktop.DBus.Error.NoReply": raise self._action_done(self.ACTION_PRE_INSTALL, authorized=False, success=False, error_string=None, error_desc=None) except Exception: self._action_done(self.ACTION_PRE_INSTALL, authorized=True, success=False, error_string=None, error_desc=None) raise def _update_oem(self, trans, status, action): # This is the "finished" handler of installing an oem metapackage # What we do now is: # 1. update_cache() for the new sources.lists only if status == EXIT_FAILED: self._show_transaction_error(trans, action) return (install, _, _, _, _, _) = trans.packages self._oem_packages_to_update = set(install) self._len_oem_updates = len(install) self._apt_update_oem() @inline_callbacks def commit_oem(self, pkgs_install_oem, pkgs_upgrade_oem): self.all_oem_packages = set(pkgs_install_oem) | set(pkgs_upgrade_oem) # Nothing to do? Go to the regular updates. try: if not pkgs_install_oem and not pkgs_upgrade_oem: self._action_done(self.ACTION_PRE_INSTALL, authorized=True, success=True, error_string=None, error_desc=None, trans_failed=None) return if pkgs_install_oem: trans = yield self.client.install_packages(pkgs_install_oem, defer=True) yield self._show_transaction( trans, self.ACTION_PRE_INSTALL, _("Installing updates…"), True, on_finished_handler=self._update_oem, progress_bar_max=0.1 ) except errors.NotAuthorizedError: self._action_done(self.ACTION_PRE_INSTALL, authorized=False, success=False, error_string=None, error_desc=None) except errors.TransactionFailed as e: self.trans_failed_msg = str(e) except dbus.DBusException as e: if e.get_dbus_name() != "org.freedesktop.DBus.Error.NoReply": raise self._action_done(self.ACTION_PRE_INSTALL, authorized=False, success=False, error_string=None, error_desc=None) except Exception: self._action_done(self.ACTION_PRE_INSTALL, authorized=True, success=False, error_string=None, error_desc=None) raise @inline_callbacks def commit(self, pkgs_install, pkgs_upgrade, pkgs_remove): """Commit a list of package adds and removes""" try: reinstall = purge = downgrade = [] trans = yield self.client.commit_packages( pkgs_install, reinstall, pkgs_remove, purge, pkgs_upgrade, downgrade, defer=True) yield self._show_transaction(trans, self.ACTION_INSTALL, _("Installing updates…"), True) except errors.NotAuthorizedError: self._action_done(self.ACTION_INSTALL, authorized=False, success=False, error_string=None, error_desc=None) except errors.TransactionFailed as e: self.trans_failed_msg = str(e) except dbus.DBusException as e: #print(e, e.get_dbus_name()) if e.get_dbus_name() != "org.freedesktop.DBus.Error.NoReply": raise self._action_done(self.ACTION_INSTALL, authorized=False, success=False, error_string=None, error_desc=None) except Exception: self._action_done(self.ACTION_INSTALL, authorized=True, success=False, error_string=None, error_desc=None) raise def _on_details_changed(self, trans, details, label_details): label_details.set_label(details) def _on_status_changed(self, trans, status, label_details): label_details.set_label(get_status_string_from_enum(status)) # Also resize the window if we switch from download details to # the terminal window if status == STATUS_COMMITTING and self._expander \ and self._expander.terminal.get_visible(): self._resize_to_show_details(self._expander) @inline_callbacks def _show_transaction(self, trans, action, header, show_details, progress_bar_max=1, on_finished_handler=None): if on_finished_handler is None: on_finished_handler = self._on_finished self.label_header.set_label(header) if not self.progressbar: self.progressbar = AptStackedProgressBar(self.unity) self.progressbar.show() self.progressbar_slot.add(self.progressbar) self.progressbar.add_transaction(trans, progress_bar_max) if self.button_cancel: self.button_cancel.set_transaction(trans) else: self.button_cancel = AptCancelButton(trans) self.button_cancel.show() self.button_cancel_slot.add(self.button_cancel) if action == self.ACTION_UPDATE: self.button_cancel.set_label(Gtk.STOCK_STOP) if show_details: if not self._expander: self._expander = UpdateManagerExpander(trans) self._expander.set_vexpand(True) self._expander.set_hexpand(True) self._expander.show_all() self._expander.connect("notify::expanded", self._on_expanded) self.expander_slot.add(self._expander) self.expander_slot.show() else: self._expander.set_transaction(trans) self._expander.final = action != self.ACTION_PRE_INSTALL elif self._expander: self._expander_slot.hide() trans.connect("status-details-changed", self._on_details_changed, self.label_details) trans.connect("status-changed", self._on_status_changed, self.label_details) trans.connect("finished", on_finished_handler, action) trans.connect("medium-required", self._on_medium_required) trans.connect("config-file-conflict", self._on_config_file_conflict) yield trans.set_debconf_frontend("gnome") yield trans.run() def _on_expanded(self, expander, param): # Make the dialog resizable if the expander is expanded # try to restore a previous size if not expander.get_expanded(): self._expanded_size = (expander.terminal.get_visible(), self.window_main.get_size()) self.window_main.end_user_resizable() elif self._expanded_size: term_visible, (stored_width, stored_height) = self._expanded_size # Check if the stored size was for the download details or # the terminal widget if term_visible != expander.terminal.get_visible(): # The stored size was for the download details, so we need # get a new size for the terminal widget self._resize_to_show_details(expander) else: self.window_main.begin_user_resizable(stored_width, stored_height) else: self._resize_to_show_details(expander) def _resize_to_show_details(self, expander): """Resize the window to show the expanded details. Unfortunately the expander only expands to the preferred size of the child widget (e.g showing all 80x24 chars of the Vte terminal) if the window is rendered the first time and the terminal is also visible. If the expander is expanded afterwards the window won't change its size anymore. So we have to do this manually. See LP#840942 """ if expander.get_expanded(): win_width, win_height = self.window_main.get_size() exp_width = expander.get_allocation().width exp_height = expander.get_allocation().height if expander.terminal.get_visible(): terminal_width = expander.terminal.get_char_width() * 80 terminal_height = expander.terminal.get_char_height() * 24 new_width = terminal_width - exp_width + win_width new_height = terminal_height - exp_height + win_height else: new_width = win_width + 100 new_height = win_height + 200 self.window_main.begin_user_resizable(new_width, new_height) def _on_medium_required(self, transaction, medium, drive): dialog = AptMediumRequiredDialog(medium, drive, self.window_main) res = dialog.run() dialog.hide() if res == Gtk.ResponseType.OK: transaction.provide_medium(medium) else: transaction.cancel() def _on_config_file_conflict(self, transaction, old, new): dialog = AptConfigFileConflictDialog(old, new, self.window_main) res = dialog.run() dialog.hide() if res == Gtk.ResponseType.YES: transaction.resolve_config_file_conflict(old, "replace") else: transaction.resolve_config_file_conflict(old, "keep") def _on_finished(self, trans, status, action): error_string = None error_desc = None trans_failed = False if status == EXIT_FAILED: error_string = get_error_string_from_enum(trans.error.code) error_desc = get_error_description_from_enum(trans.error.code) if self.trans_failed_msg: trans_failed = True error_desc = error_desc + "\n" + self.trans_failed_msg # tell unity to hide the progress again self.unity.set_progress(-1) is_success = (status == EXIT_SUCCESS) try: self._action_done(action, authorized=True, success=is_success, error_string=error_string, error_desc=error_desc, trans_failed=trans_failed) except TypeError: # this module used to be be lazily imported and in older code # trans_failed= is not accepted # TODO: this workaround can be dropped in Ubuntu 20.10 self._action_done(action, authorized=True, success=is_success, error_string=error_string, error_desc=error_desc) if __name__ == "__main__": import mock options = mock.Mock() data_dir = "/usr/share/update-manager" from UpdateManager.UpdateManager import UpdateManager app = UpdateManager(data_dir, options) b = InstallBackendAptdaemon(app, None) b.commit(["2vcard"], [], []) Gtk.main() backend/InstallBackendSynaptic.py 0000644 00000005603 15030301074 0013073 0 ustar 00 #!/usr/bin/env python # -*- Mode: Python; indent-tabs-mode: nil; tab-width: 4; coding: utf-8 -*- # (c) 2005-2007 Canonical, GPL import os import tempfile from gettext import gettext as _ import gi gi.require_version("Gtk", "3.0") from gi.repository import GObject # Extra GdkX11 import for pygobject bug #673396 # https://bugzilla.gnome.org/show_bug.cgi?id=673396 from gi.repository import GdkX11 GdkX11 # pyflakes from UpdateManager.backend import InstallBackend class InstallBackendSynaptic(InstallBackend): """ Install backend based on synaptic """ def update(self): opt = ["--update-at-startup"] tempf = None self._run_synaptic(self.ACTION_UPDATE, opt, tempf) def commit(self, pkgs_install, pkgs_upgrade, pkgs_remove, close_on_done=False): # close when update was successful (its ok to use a Synaptic:: # option here, it will not get auto-saved, because synaptic does # not save options in non-interactive mode) opt = [] if close_on_done: opt.append("-o") opt.append("Synaptic::closeZvt=true") # custom progress strings opt.append("--progress-str") opt.append("%s" % _("Please wait, this can take some time.")) opt.append("--finish-str") opt.append("%s" % _("Update is complete")) tempf = tempfile.NamedTemporaryFile(mode="w+") for pkg_name in pkgs_install + pkgs_upgrade: tempf.write("%s\tinstall\n" % pkg_name) for pkg_name in pkgs_remove: tempf.write("%s\tdeinstall\n" % pkg_name) opt.append("--set-selections-file") opt.append("%s" % tempf.name) tempf.flush() self._run_synaptic(self.ACTION_INSTALL, opt, tempf) def _run_synaptic(self, action, opt, tempf): """Execute synaptic.""" win = self.window_main.get_window() try: xid = win.get_xid() except AttributeError: xid = 0 cmd = ["/usr/bin/pkexec", "/usr/sbin/synaptic", "--hide-main-window", "--non-interactive", "--parent-window-id", "%s" % xid] cmd.extend(opt) flags = GObject.SPAWN_DO_NOT_REAP_CHILD (pid, stdin, stdout, stderr) = GObject.spawn_async(cmd, flags=flags) # Keep a reference to the data tuple passed to # GObject.child_watch_add to avoid attempts to destroy it without a # thread context: https://bugs.launchpad.net/bugs/724687 self.child_data = (action, tempf) GObject.child_watch_add(pid, self._on_synaptic_exit, self.child_data) def _on_synaptic_exit(self, pid, condition, data): action, tempf = data if tempf: tempf.close() self._action_done(action, authorized=True, success=os.WEXITSTATUS(condition) == 0, error_string=None, error_desc=None) backend/__init__.py 0000644 00000040324 15030301074 0010240 0 ustar 00 #!/usr/bin/env python # -*- Mode: Python; indent-tabs-mode: nil; tab-width: 4; coding: utf-8 -*- """Integration of package managers into UpdateManager""" # (c) 2005-2009 Canonical, GPL from __future__ import absolute_import import gi gi.require_version("Gtk", "3.0") gi.require_version('Snapd', '1') from gi.repository import GLib, Gtk, Snapd from apt import Cache import json import logging import os import re import subprocess from gettext import gettext as _ from threading import Thread from UpdateManager.Core.MyCache import MyCache from UpdateManager.Core.utils import inhibit_sleep, get_dist_version from UpdateManager.Dialogs import Dialog class InstallBackend(Dialog): ACTION_UPDATE = 0 ACTION_PRE_INSTALL = 1 ACTION_INSTALL = 2 def __init__(self, window_main, action): Dialog.__init__(self, window_main) self.action = action self.sleep_cookie = None def start(self): os.environ["APT_LISTCHANGES_FRONTEND"] = "none" # Do not suspend during the update process self.sleep_cookie = inhibit_sleep() if self.action == self.ACTION_PRE_INSTALL: unfresh_cache = self.window_main.cache fresh_cache = Cache(rootdir=self.window_main.cache.rootdir) # Install OEM packages, update, then do ACTION_INSTALL pkgs_install_oem = [] pkgs_upgrade_oem = [] for pkg in self.window_main.oem_metapackages: unfresh_pkg = unfresh_cache[pkg] fresh_pkg = fresh_cache[pkg] if (unfresh_pkg.marked_install and not fresh_pkg.is_installed): pkgs_install_oem.append(pkg) elif (unfresh_pkg.marked_upgrade and fresh_pkg.is_upgradable): pkgs_upgrade_oem.append(pkg) self.commit_oem(pkgs_install_oem, pkgs_upgrade_oem) elif self.action == self.ACTION_INSTALL: # Get the packages which should be installed and update pkgs_install = [] pkgs_upgrade = [] pkgs_remove = [] # Get a fresh cache in case update-manager's is outdated to # skip operations that already took place fresh_cache = Cache(rootdir=self.window_main.cache.rootdir) for pkg in self.window_main.cache: try: if pkg.marked_install \ and not fresh_cache[pkg.name].is_installed: pkgname = pkg.name if pkg.is_auto_installed: pkgname += "#auto" pkgs_install.append(pkgname) elif (pkg.marked_upgrade and fresh_cache[pkg.name].is_upgradable): pkgs_upgrade.append(pkg.name) elif (pkg.marked_delete and fresh_cache[pkg.name].is_installed): pkgs_remove.append(pkg.name) except KeyError: # pkg missing from fresh_cache can't be modified pass self.commit(pkgs_install, pkgs_upgrade, pkgs_remove) else: self.update() def update(self): """Run a update to refresh the package list""" raise NotImplementedError def commit_oem(self, pkgs_install_oem, pkgs_upgrade_oem): """ Install these OEM packages """ self._action_done(self.ACTION_PRE_INSTALL, authorized=True, success=True, error_string=None, error_desc=None, trans_failed=None) def commit(self, pkgs_install, pkgs_upgrade, pkgs_remove): """Commit the cache changes """ raise NotImplementedError def get_snap_seeds(self): seeded_snaps = {} unseeded_snaps = {} curr_channel = "stable/ubuntu-" + get_dist_version() cache = self.window_main.cache try: d2s_file = open( '/usr/share/ubuntu-release-upgrader/deb2snap.json', 'r') d2s = json.load(d2s_file) d2s_file.close() for snap in d2s["seeded"]: seed = d2s["seeded"][snap] deb = seed.get("deb", None) to_channel = seed.get("to_channel", curr_channel) metapkg = seed.get("metapkg", None) if metapkg not in cache: continue if metapkg and cache[metapkg].is_installed is False: continue seeded_snaps[snap] = (deb, to_channel) for snap in d2s["unseeded"]: unseed = d2s["unseeded"][snap] from_channel = unseed.get("from_channel", curr_channel) metapkg = seed.get("metapkg", None) if metapkg not in cache: continue if metapkg and cache[metapkg].is_installed is False: continue unseeded_snaps[snap] = (from_channel) except Exception as e: logging.debug("error reading deb2snap.json file (%s)" % e) return seeded_snaps, unseeded_snaps def get_deb2snap_dups(self): # update and grab the latest cache try: if self.window_main.cache is None: self.window_main.cache = MyCache(None) else: self.window_main.cache.open(None) self.window_main.cache._initDepCache() cache = self.window_main.cache except Exception as e: # just return an empty array for now, it's perfectly safe to # postpone this duplicates check to a later update. logging.debug("error reading cache (%s)" % e) return [] duplicates = [] seeded_snaps, _ = self.get_snap_seeds() for snap, (deb, _) in seeded_snaps.items(): # if the deb is installed and was not manually installed, # replace it if (deb in cache and cache[deb].is_installed): deb_is_auto = True cache[deb].mark_delete() for pkg in cache.get_changes(): if (pkg.is_installed and pkg.marked_delete and not pkg.is_auto_installed): deb_is_auto = False break cache.clear() if deb_is_auto: duplicates.append(deb) return duplicates def get_snap_transitions(self): # populate snap_list with deb2snap transitions snap_list = {} seeded_snaps, unseeded_snaps = self.get_snap_seeds() for snap, (deb, to_channel) in seeded_snaps.items(): snap_object = {} # check if the snap is already installed snap_info = subprocess.Popen(["snap", "info", snap], universal_newlines=True, stdout=subprocess.PIPE).communicate() if re.search("^installed: ", snap_info[0], re.MULTILINE): logging.debug("Snap %s is installed" % snap) continue elif (deb in self.window_main.duplicate_packages): # install the snap if the deb was just marked delete snap_object['command'] = 'install' snap_object['channel'] = to_channel snap_list[snap] = snap_object for snap, (from_channel) in unseeded_snaps.items(): snap_object = {} # check if the snap is already installed snap_info = subprocess.Popen(["snap", "info", snap], universal_newlines=True, stdout=subprocess.PIPE).communicate() if re.search("^installed: ", snap_info[0], re.MULTILINE): logging.debug("Snap %s is installed" % snap) # its not tracking the release channel so don't remove re_channel = "stable/ubuntu-[0-9][0-9].[0-9][0-9]" if not re.search(r"^tracking:.*%s" % re_channel, snap_info[0], re.MULTILINE): logging.debug("Snap %s is not tracking the release channel" % snap) continue snap_object['command'] = 'remove' # check if this snap is being used by any other snaps conns = subprocess.Popen(["snap", "connections", snap], universal_newlines=True, stdout=subprocess.PIPE).communicate() for conn in conns[0].split('\n'): conn_cols = conn.split() if len(conn_cols) != 4: continue plug = conn_cols[1] slot = conn_cols[2] if slot.startswith(snap + ':'): plug_snap = plug.split(':')[0] if plug_snap != '-' and \ plug_snap not in unseeded_snaps: logging.debug("Snap %s is being used by %s. " "Switching it to stable track" % (snap, plug_snap)) snap_object['command'] = 'refresh' snap_object['channel'] = 'stable' break snap_list[snap] = snap_object return snap_list def update_snap_cb(self, client, change, _, user_data): index, count, progress_bar = user_data if not progress_bar: return # determine how much of this change has been done task_total = 0 task_done = 0 for task in change.get_tasks(): task_total += task.get_progress_total() task_done += task.get_progress_done() task_fraction = task_done / task_total # determine how much total progress has been made total_fraction = (task_fraction / count) + (index / count) # change.get_tasks() can increase between callbacks so we must # avoid jumping backward in progress here if total_fraction > progress_bar.get_fraction(): GLib.idle_add(progress_bar.set_fraction, total_fraction) def update_snaps(self): # update status and progress bar def update_status(status): GLib.idle_add(self.label_details.set_label, status) def update_progress(progress_bar): progress_bar.pulse() return True update_status(_("Updating snaps")) progress_bar = None progress_timer = None progress_bars = self.progressbar_slot.get_children() if progress_bars and isinstance(progress_bars[0], Gtk.ProgressBar): progress_bar = progress_bars[0] progress_timer = GLib.timeout_add(100, update_progress, progress_bar) # populate snap_list with deb2snap transitions snap_list = self.get_snap_transitions() if progress_timer: GLib.source_remove(progress_timer) progress_bar.set_fraction(0) # (un)install (un)seeded snap(s) try: client = Snapd.Client() client.connect_sync() index = 0 count = len(snap_list) for snap, snap_object in snap_list.items(): command = snap_object['command'] if command == 'refresh': update_status(_("Refreshing %s snap" % snap)) client.refresh_sync(snap, snap_object['channel'], self.update_snap_cb, progress_callback_data=(index, count, progress_bar)) elif command == 'remove': update_status(_("Removing %s snap" % snap)) client.remove_sync(snap, self.update_snap_cb, progress_callback_data=(index, count, progress_bar)) else: update_status(_("Installing %s snap" % snap)) client.install_sync(snap, snap_object['channel'], self.update_snap_cb, progress_callback_data=(index, count, progress_bar)) index += 1 except GLib.Error as e: logging.debug("error updating snaps (%s)" % e) GLib.idle_add(self.window_main.start_error, False, _("Upgrade only partially completed."), _("An error occurred while updating snaps. " "Please check your network connection.")) return # continue with the rest of the updates GLib.idle_add(self.window_main.start_available) def _action_done(self, action, authorized, success, error_string, error_desc, trans_failed=False): # If the progress dialog should be closed automatically afterwards #settings = Gio.Settings.new("com.ubuntu.update-manager") #close_after_install = settings.get_boolean( # "autoclose-install-window") # FIXME: confirm with mpt whether this should still be a setting #close_after_install = False if action == self.ACTION_PRE_INSTALL and success: # Now do the regular updates self.action = self.ACTION_INSTALL self.start() elif action == self.ACTION_INSTALL: if (success and os.path.exists("/usr/bin/snap") and hasattr(self, 'pane_update_progress')): Thread(target=self.update_snaps).start() elif success: self.window_main.start_available() elif error_string: self.window_main.start_error(trans_failed, error_string, error_desc) else: # exit gracefuly, we can't just exit as this will trigger # a crash if system.exit() is called in a exception handler GLib.timeout_add(1, self.window_main.exit) else: if error_string: self.window_main.start_error(True, error_string, error_desc) elif (success and os.path.exists("/usr/bin/snap") and hasattr(self, 'pane_update_progress')): self.window_main.duplicate_packages = self.get_deb2snap_dups() self.window_main.start_available() else: is_cancelled_update = not success self.window_main.start_available(is_cancelled_update) # try aptdaemon if os.path.exists("/usr/sbin/aptd") \ and "UPDATE_MANAGER_FORCE_BACKEND_SYNAPTIC" not in os.environ: # check if the gtkwidgets are installed as well try: from .InstallBackendAptdaemon import InstallBackendAptdaemon except ImportError: logging.exception("importing aptdaemon") # try synaptic if os.path.exists("/usr/sbin/synaptic") \ and "UPDATE_MANAGER_FORCE_BACKEND_APTDAEMON" not in os.environ: try: from .InstallBackendSynaptic import InstallBackendSynaptic except ImportError: logging.exception("importing synaptic") def get_backend(*args, **kwargs): """Select and return a package manager backend.""" # try aptdaemon if (os.path.exists("/usr/sbin/aptd") and "UPDATE_MANAGER_FORCE_BACKEND_SYNAPTIC" not in os.environ): # check if the gtkwidgets are installed as well try: return InstallBackendAptdaemon(*args, **kwargs) except NameError: logging.exception("using aptdaemon failed") # try synaptic if (os.path.exists("/usr/sbin/synaptic") and "UPDATE_MANAGER_FORCE_BACKEND_APTDAEMON" not in os.environ): try: return InstallBackendSynaptic(*args, **kwargs) except NameError: pass # nothing found, raise raise Exception("No working backend found, please try installing " "aptdaemon or synaptic") __init__.py 0000644 00000000000 15030301074 0006634 0 ustar 00
| ver. 1.4 |
Github
|
.
| PHP 8.2.28 | Generation time: 0.02 |
proxy
|
phpinfo
|
Settings