common module

exception common.EWaptCancelled[source]

Bases: exceptions.Exception

class common.LogInstallOutput(console, waptdb, rowid)[source]

Bases: object

file like to log print output to db installstatus


alias of package

class common.SYSTEM_POWER_STATUS[source]

Bases: _ctypes.Structure


Structure/Union member


Structure/Union member


Structure/Union member


Structure/Union member


Structure/Union member


Structure/Union member



class common.Wapt(config=None, config_filename=None, defaults=None, disable_update_server_status=True)[source]

Bases: object

Global WAPT engine


Add an automatic host repository, remove existing WaptHostRepo last one before

add_iconpng_wapt(package, iconpath='', private_key_passwd=None)[source]

Add a WAPT/icon.png file to existing WAPT package without icon #>>> wapt = Wapt(config_filename=’c:/wapt/wapt-get.ini’) #>>> res = wapt.add_iconpng_wapt(‘tis-firefox’)


Add a pyscripter project file to package development directory.

Parameters:target_directory (str) – path to location where to create the wa^t.psproj file.

Add a local shitdown policy to upgrade system

build_package(directoryname, inc_package_release=False, excludes=['.svn', '.git', '.gitignore', '*.pyc', 'src'], target_directory=None)[source]

Build the WAPT package from a directory

Call update_control from if this function is defined. Then zip the content of directory. Add a manifest.sha1 file with sha1 hash of

the content of each file.
  • directoryname (str) – source root directory of package to build
  • inc_package_release (boolean) – increment the version of package in control file.

{‘filename’:waptfilename,’files’:[list of files],’package’:PackageEntry}

Return type:


build_upload(sources_directories, private_key_passwd=None, wapt_server_user=None, wapt_server_passwd=None, inc_package_release=False, target_directory=None)[source]
Build a list of packages and upload the resulting packages to the main repository.
if section of package is group or host, user specific wapt-host or wapt-group
Returns:{‘filename’:waptfilename,’files’:[list of files],’package’:PackageEntry}
Return type:list of build result dict
check_cancelled(msg='Task cancelled')[source]
check_depends(apackages, forceupgrade=False, force=False, assume_removed=[])[source]

Given a list of packagename or requirement “name (=version)”, return a dictionnary of {‘additional’ ‘upgrade’ ‘install’ ‘skipped’ ‘unavailable’,’remove’} of [packagerequest,matching PackageEntry]

  • apackages (str or list) – list of packages for which to check missing dependencies.
  • forceupgrade (boolean) – if True, check if the current installed packages is the latest available
  • force (boolean) – if True, install the latest version even if the package is already there and match the requirement
  • assume_removed (list) – list of packagename which are assumed to be absent even if they are installed to check the consequences of removal of packages, implies force=True

{‘additional’ ‘upgrade’ ‘install’ ‘skipped’ ‘unavailable’, ‘remove’} with list of [packagerequest,matching PackageEntry]

Return type:



Return list of available package entries not yet in cache to match supplied packages requirements


Check and return the host package if available and not installed

check_install(apackages=None, force=True, forceupgrade=True)[source]

Return a list of actions required for install of apackages list of packages if apackages is None, check for all pending updates.

  • apackages (str or list) – list of packages or None to check pending install/upgrades
  • force (boolean) – if True, already installed package listed in apackages will be considered to be reinstalled
  • forceupgrade – if True, all dependencies are upgraded to latest version, even if current version comply with depends requirements

with keys [‘skipped’, ‘additional’, ‘remove’, ‘upgrade’, ‘install’, ‘unavailable’] and list of (package requirements, PackageEntry)

Return type:



Check if an install is in progress, return list of pids of install in progress Kill old stucked wapt-get processes/children and update db status max_ttl is maximum age of wapt-get in minutes


Return a list of additional package to remove if apackages are removed

Parameters:apackages (str or list) – list of packages fr which parent dependencies will be checked.
Returns:list of package requirements with broken dependencies
Return type:list

Check if a new version of the Wapt client is available return url and version if newer return None if no update


Remove cached WAPT files from local disk

Parameters:obsolete_only (boolean) – If True, remove packages which are either no more available, or installed at a equal or newer version
Returns:list of filenames of removed packages
Return type:list
>>> wapt = Wapt(config_filename='c:/wapt/wapt-get.ini')
>>> l = wapt.download_packages(wapt.check_downloads())
>>> res = wapt.cleanup(True)

Remove all current user session_setup informations for removed packages

corrupted_files_sha1(rootdir, manifest)[source]

check hexdigest sha1 for the files in manifest returns a list of non matching files (corrupted files)


Remove a key from local db

dependencies(packagename, expand=False)[source]

Return all dependecies of a given package >>> w = Wapt(config_filename=’c:/wapt/wapt-get.ini’) >>> dep = w.dependencies(‘tis-waptdev’) >>> isinstance(dep,list) and isinstance(dep[0],PackageEntry) True


Disable Wapt automatic update/upgrade scheduling through windows scheduler

download_packages(package_requests, usecache=True, printhook=None)[source]

Download a list of packages (requests are of the form packagename (>version) ) returns a dict of {“downloaded,”skipped”,”errors”}

  • package_requests (str or list) – list of packages to prefetch
  • usecache (boolean) – if True, don’t download package if already in cache
  • printhook (func) – callback with signature report(received,total,speed,url) to display progress

with keys {“downloaded,”skipped”,”errors”} and list of cached file paths.

Return type:


>>> wapt = Wapt(config_filename='c:/wapt/wapt-get.ini')
>>> def nullhook(*args):
...     pass
>>> wapt.download_packages(['tis-firefox','tis-waptdev'],usecache=False,printhook=nullhook)
{'downloaded': [u'c:/wapt\\cache\\tis-firefox_37.0.2-9_all.wapt', u'c:/wapt\\cache\\tis-waptdev.wapt'], 'skipped': [], 'errors': []}

Download packages that can be upgraded

duplicate_package(packagename, newname=None, newversion=None, target_directory=None, build=True, excludes=['.svn', '.git', '.gitignore', '*.pyc', 'src'], private_key=None, callback=<function pwd_callback at 0x023FD570>, append_depends=None, remove_depends=None, append_conflicts=None, remove_conflicts=None, auto_inc_version=True, usecache=True, printhook=None)[source]

Duplicate an existing package.

Duplicate an existing package from declared repostory or file into targetdirectory with
optional newname and version.
  • newname (str) – name of target package
  • newversion (str) – version of target package. if None, use source package version
  • target_directory (str) – path where to put development files. If None, use temporary. If empty, use default development dir
  • build (bool) – If True, build and sign the package. The filename of build package will be in ‘target’ key of result
  • callback (func) – function to get rawbytes password of private key
  • append_depends (list) – comma str or list of depends to append.
  • remove_depends (list) – comma str or list of depends to remove.
  • auto_inc_version (bool) – if version is less than existing package in repo, set version to repo version+1
  • usecache (bool) – If True, allow to use cached package in local repo instead of downloading it.
  • printhook (func) – hook for download progress

{‘target’:new package if build, or ‘source_dir’:new source directory if not build ,’package’:new PackageEntry}

Return type:


>>> wapt = Wapt(config_filename='c:/tranquilit/wapt/tests/wapt-get.ini')
>>> wapt.dbpath = ':memory:'
>>> r= wapt.update()
>>> def nullhook(*args):
...     pass
>>> tmpdir = 'c:/tmp/testdup-wapt'
>>> if os.path.isdir(tmpdir):
...     import shutil
...     shutil.rmtree(tmpdir)
>>> p = wapt.duplicate_package('tis-wapttest',
...     newname='testdup',
...     newversion='20.0-0',
...     target_directory=tmpdir,
...     build=False,
...     excludes=['.svn','.git','.gitignore','*.pyc','src'],
...     private_key=None,
...     callback=pwd_callback,
...     append_depends=None,
...     auto_inc_version=True,
...     usecache=False,
...     printhook=nullhook)
>>> print repr(p['package'])
>>> if os.path.isdir(tmpdir):
...     import shutil
...     shutil.rmtree(tmpdir)
>>> p = wapt.duplicate_package('tis-wapttest',
...    target_directory=tempfile.mkdtemp('wapt'),
...    build=True,
...    auto_inc_version=True,
...    append_depends=['tis-firefox','tis-irfanview'],
...    remove_depends=['tis-wapttestsub'],
...    )
>>> print repr(p['package'])
edit_host(hostname, target_directory='', use_local_sources=True, append_depends=None, remove_depends=None, append_conflicts=None, remove_conflicts=None, printhook=None, description=None)[source]
Download and extract a host package from host repositories into target_directory for modification
Return dict {‘target’: ‘c:\tmp\dummy’, ‘source_dir’: ‘c:\tmp\dummy’, ‘package’: “dummy.tranquilit.local (=0)”}

hostname : fqdn of the host to edit target_directory : where to place the developments files. if empty, use default one from wapt-get.ini configuration use_local_sources : don’t raise an exception if local sources are newer or same than repo version append_depends : list or comma separated list of package requirements remove_depends : list or comma separated list of package requirements to remove

>>> wapt = Wapt(config_filename='c:/wapt/wapt-get.ini')
>>> tmpdir = 'c:/tmp/dummy'
>>> wapt.edit_host('dummy.tranquilit.local',target_directory=tmpdir,append_depends='tis-firefox')
{'target': 'c:\\tmp\\dummy', 'source_dir': 'c:\\tmp\\dummy', 'package': PackageEntry('dummy.tranquilit.local','0')}
>>> import shutil
>>> shutil.rmtree(tmpdir)
>>> host = wapt.edit_host('htlaptop.tranquilit.local',target_directory=tmpdir,append_depends='tis-firefox')
>>> 'package' in host
>>> shutil.rmtree(tmpdir)
edit_package(packagerequest, target_directory='', use_local_sources=True, append_depends=None, remove_depends=None, append_conflicts=None, remove_conflicts=None, auto_inc_version=True)[source]
Download an existing package from repositories into target_directory for modification
if use_local_sources is True and no newer package exists on repos, updates current local edited data
else if target_directory exists and is not empty, raise an exception

Return {‘target’:target_directory,’source_dir’:target_directory,’package’:package_entry}

  • packagerequest (str) – path to existing wapt file, or package request
  • use_local_sources (boolean) – don’t raise an exception if target exist and match package version


Return type:


>>> wapt = Wapt(config_filename='c:/tranquilit/wapt/tests/wapt-get.ini')
>>> wapt.dbpath = ':memory:'
>>> r= wapt.update()
>>> tmpdir = tempfile.mkdtemp('wapt')
>>> res = wapt.edit_package('tis-wapttest',target_directory=tmpdir,append_depends='tis-firefox',remove_depends='tis-7zip')
>>> res['target'] == tmpdir and res['package'].package == 'tis-wapttest' and 'tis-firefox' in res['package'].depends
>>> import shutil
>>> shutil.rmtree(tmpdir)

Enable Wapt automatic update/upgrade scheduling through windows scheduler


return install tasks with error status

Remove install status for packages from local database
without actually uninstalling the packages
Parameters:packages_list (list) – list of installed package names to forget
Returns:list of package names actually forgotten
Return type:list
>>> wapt = Wapt(config_filename='c:/wapt/wapt-get.ini')
>>> res = wapt.install('tis-test')
>>> res = wapt.is_installed('tis-test')
>>> isinstance(res,PackageEntry)
>>> wapt.forget_packages('tis-test')
>>> wapt.is_installed('tis-test')
>>> print wapt.is_installed('tis-test')

Regenerate a random UUID for this host or force with supplied one.

forced_uuid (str): uuid to force for this host. If None, generate a random one

Normally, the UUID is taken from BIOS through wmi.

In case bios returns some duplicates or garbage, it can be useful to
force a random uuid. This is stored as uuid key in wapt-get.ini.
In case we want to link th host with a an existing record on server, we
can force a old UUID.
get_default_development_dir(packagecond, section='base')[source]
Returns the default developement directory for package named <packagecond>
based on default_sources_root and default_sources_suffix ini parameters
Parameters:packagecond (str) – packahe name or pacjage request “name (=version)”
Returns:path to local development directory
Return type:str

Return most up to date packages entries for packages_names packages_names is either a list or a string return a dictionnary with {‘packages’:[],’missing’:[]} >>> wapt = Wapt(config_filename=’c:/wapt/wapt-get.ini’) >>> res = wapt.get_package_entries([‘tis-firefox’,’tis-putty’]) >>> isinstance(res[‘missing’],list) and isinstance(res[‘packages’][0],PackageEntry) True


Download sources of package (if referenced in package as a https svn) in the current directory

global_attributes = ['wapt_base_dir', 'waptserver', 'config_filename', 'proxies', 'repositories', 'private_key', 'public_certs', 'package_cache_dir', 'dbpath']

Creates a set of private/public keys to identify the host on the waptserver keys are created in the same directory as DB


Return package name for current computer

http_upload_package(package, wapt_server_user=None, wapt_server_passwd=None)[source]
Upload a package or host package to the waptserver.
package : either the filename of a wapt package, or a PackageEntry wapt_server_user : wapt_server_passwd :
>>> from common import *
>>> wapt = Wapt(config_filename = r'C:\tranquilit\wapt\tests\wapt-get.ini')
>>> r = wapt.update()
>>> d = wapt.duplicate_package('tis-wapttest','toto')
>>> print d
{'target': u'c:\\users\\htouvet\\appdata\\local\\temp\\toto.wapt', 'package': PackageEntry('toto','119')}
>>> wapt.http_upload_package(d['package'],wapt_server_user='admin',wapt_server_passwd='password')
install(apackages, force=False, params_dict={}, download_only=False, usecache=True, printhook=None)[source]

Install a list of packages and its dependencies removes first packages which are in conflicts package attribute Returns a dictionary of (package requirement,package) with ‘install’,’skipped’,’additional’

  • apackages – list of packages requirements “packagename(=version)” or list of PackageEntry.
  • force – reinstalls the packages even if it is already installed
  • params_dict – dict of parameters passed to the install() procedure in the packages of all packages as params variables and as “setup module” attributes
  • download_only – don’t install package, but only download them
  • usecache – use the already downloaded packages if available in cache directory
  • printhook – hook for progress print

with keys [‘skipped’, ‘additional’, ‘remove’, ‘upgrade’, ‘install’, ‘unavailable’] and list of (package requirements, PackageEntry)

Return type:


>>> wapt = Wapt(config_filename='c:/tranquilit/wapt/tests/wapt-get.ini')
>>> def nullhook(*args):
...     pass
>>> res = wapt.install(['tis-wapttest'],usecache=False,printhook=nullhook,params_dict=dict(company='toto'))
>>> isinstance(res['upgrade'],list) and isinstance(res['errors'],list) and isinstance(res['additional'],list) and isinstance(res['install'],list) and isinstance(res['unavailable'],list)
>>> res = wapt.remove('tis-wapttest')
>>> res == {'removed': ['tis-wapttest'], 'errors': []}
install_wapt(fname, params_dict={}, explicit_by=None)[source]

Install a single wapt package given its WAPT filename. return install status


Returns all installed packages with their status

Parameters:include_errors (boolean) – include packages wnot installed successfully
Returns:list of PackageEntry merged with local install status.
Return type:list

Return full inventory of the computer as a dictionary.

Return type:dict
Check if a package (with optional version condition) is available
in repositories. Return list of matching package entries or empty list
>>> wapt = Wapt(config_filename='c:/tranquilit/wapt/tests/wapt-get.ini')
>>> l = wapt.is_available('tis-wapttest')
>>> l and isinstance(l[0],PackageEntry)
is_installed(packagename, include_errors=False)[source]

Checks if a package is installed. Return package entry and additional local status or None

Parameters:packagename (str) – name / package request to query
Returns:None en PackageEntry merged with local install_xxx fields * install_date * install_output * install_params * install_status
Return type:PackageEntry

Return PackageEntry if directory is a wapt developement directory (a WAPT/control file exists) or False


Return PackageEntry if filename is a wapt package or False


Get the printed output of the last install of package named packagename

Parameters:packagename (str) – name of package to query
Returns:{status,log} of the last install of a package
Return type:dict
>>> w = Wapt()
>>> w.last_install_log('tis-7zip')
{'status': u'OK', 'log': u'Installing 7-Zip 9.38.0-1\n7-Zip already installed, skipping msi install\n'}

Returns a list of installed packages which have the searchwords in their description


Returns a list of packages requirement which can be upgraded

Returns:{‘additional’: [], ‘install’: [], ‘remove’: [], ‘upgrade’: []}
Return type:dict

Load configuration parameters from supplied inifilename

make_group_template(packagename='', depends=None, directoryname=None, section='group', description=None)[source]
Build a skeleton of WAPT group package
depends : list of package dependencies.

Return the path of the skeleton

>>> wapt = Wapt(config_filename='c:/wapt/wapt-get.ini')
>>> tmpdir = 'c:/tmp/dummy'
>>> if os.path.isdir(tmpdir):
...    import shutil
...    shutil.rmtree(tmpdir)
>>> p = wapt.make_group_template(packagename='testgroupe',directoryname=tmpdir,depends='tis-firefox',description=u'Test de groupe')
>>> print p
{'target': 'c:\\tmp\\dummy', 'source_dir': 'c:\\tmp\\dummy', 'package': PackageEntry('testgroupe','0')}
>>> print p['package'].depends
>>> import shutil
>>> shutil.rmtree(tmpdir)
make_host_template(packagename='', depends=None, directoryname=None, description=None)[source]
make_package_template(installer_path, packagename='', directoryname='', section='', description='', depends='')[source]
Build a skeleton of WAPT package based on the properties of the supplied installer
Return the path of the skeleton
>>> wapt = Wapt(config_filename='c:/wapt/wapt-get.ini')
>>> wapt.dbpath = ':memory:'
>>> files = 'c:/tmp/files'
>>> if not os.path.isdir(files):
...    os.makedirs(files)
>>> tmpdir = 'c:/tmp/dummy'
>>> devdir = wapt.make_package_template(files,packagename='mydummy',directoryname=tmpdir,depends='tis-firefox')
>>> os.path.isfile(os.path.join(devdir,'WAPT','control'))
>>> p = wapt.build_package(devdir)
>>> 'filename' in p and isinstance(p['files'],list) and isinstance(p['package'],PackageEntry)
>>> import shutil
>>> shutil.rmtree(tmpdir)

Called whenever the network configuration has changed

packages_add_depends(packages, append_depends)[source]

Add a list of dependencies to existing packages, inc version and build-upload packages : list of package names append_depends : list of dependencies packages


Return the local IP which is most probably reachable by wapt server

In case there are several network connections, returns the local IP
which Windows choose for sending packets to WaptServer.

This can be the most probable IP which would get packets from WaptServer.

Returns:Local IP
Return type:str
read_param(name, default=None)[source]

read a param value from local db >>> wapt = Wapt(config_filename=’c:/wapt/wapt-get.ini’) >>> wapt.read_param(‘db_version’) u‘20140410’

Send computer informations to WAPT Server
if description is provided, updates local registry with new description
>>> wapt = Wapt()
>>> s = wapt.register_computer()

Return list of uninstall ID from registry launched before and after an installation to capture uninstallkey


Check if config file has been updated, Return None if config has not changed or date of new config file if reloaded >>> wapt = Wapt(config_filename=’c:/wapt/wapt-get.ini’) >>> wapt.reload_config_if_updated()

remove(packages_list, force=False)[source]

Removes a package giving its package name, unregister from local status DB

  • packages_list (str or list or path) – packages to remove (package name, list of package requirement, package entry or development directory)
  • force – if True, unregister package from local status database, even if uninstall has failed

{‘errors’: [], ‘removed’: []}

Return type:



Add a local shitdown policy to upgrade system


Reset host uuid to bios provided UUID.


Remove the local waptservice auth from ini file

run(*arg, **args)[source]
run_notfatal(*cmd, **args)[source]

Runs the command and wait for it termination returns output, don’t raise exception if exitcode is not null but return ‘’


return current install tasks


returns the current run status for tray display

search(searchwords=[], exclude_host_repo=True, section_filter=None, newest_only=False)[source]

Returns a list of packages which have the searchwords in their description

  • searchwords (str or list) – words to search in packages name or description
  • exclude_host_repo (boolean) – if True, don’t search in host repoisitories.
  • section_filter (str or list) – restrict search to the specified package sections/categories

list of packageEntry

Return type:


session_setup(packagename, force=False)[source]

Setup the user session for a specific system wide installed package” Source from database or filename

set_local_password(user='admin', pwd='password')[source]

Set admin/password local auth for waptservice in ini file as a sha256 hex hash


Setup cron job on windows for update and download-upgrade

sign_package(zip_or_directoryname, excludes=['.svn', '.git', '.gitignore', '*.pyc', 'src'], private_key=None, callback=<function pwd_callback at 0x023FD570>)[source]
Calc the signature of the WAPT/manifest.sha1 file and put/replace it in ZIP or directory.

if directory, creates WAPT/manifest.sha1 and add it to the content of package create a WAPT/signature file and it to directory or zip file.

known issue : if zip file already contains a manifest.sha1 file, it is not removed, so there will be
2 manifest files in zip / wapt package.
  • zip_or_directoryname – filename or path for the wapt package’s content
  • excludes – list of patterns to exclude from
  • private_key – path to the proivate key to use for SSL signing.
  • callback – ref to the function to call if a password is required for opening the private key.

base64 encoded signature of manifest.sha1 file (content

Return type:



Stores in DB the current pending upgrades and running installs for query by waptservice

uninstall(packagename, params_dict={})[source]

Launch the uninstall script of an installed package” Source from database or filename


return the (quiet) command stored in registry to uninstall a software given its registry key

update(force=False, register=True)[source]

Update local database with packages definition from repositories

  • force (boolean) – update even if Packages index on repository has not been updated since last update (based on http headers)
  • register (boolean) – Send informations about status of local packages to waptserver


Return type:


>>> wapt = Wapt(config_filename='c:/wapt/wapt-get.ini')
>>> updates = wapt.update()
>>> 'count' in updates and 'added' in updates and 'upgrades' in updates and 'date' in updates and 'removed' in updates

Send packages and software informations to WAPT Server, don’t send dmi

>>> wapt = Wapt()
>>> s = wapt.update_server_status()

Install “well known” host package from main repository if not already installed then query localstatus database for packages with a version older than repository and install all newest packages

{‘upgrade’: [], ‘additional’: [], ‘downloads’:
{‘downloaded’: [], ‘skipped’: [], ‘errors’: []},

‘remove’: [], ‘skipped’: [], ‘install’: [], ‘errors’: [], ‘unavailable’: []}

Return type:dict
upload_package(cmd_dict, wapt_server_user=None, wapt_server_passwd=None)[source]

Wapt configuration and version informations

Returns:versions of main main files, waptservice config, repos and waptserver config
Return type:dict
>>> w = Wapt()
>>> w.wapt_status()
        'setuphelpers-version': '1.1.1',
        'waptserver': {
                'dnsdomain': u'tranquilit.local',
                'proxies': {
                        'http': None,
                        'https': None
                'server_url': 'https: //wapt.tranquilit.local'
        'waptservice_protocol': 'http',
        'repositories': [{
                'dnsdomain': u'tranquilit.local',
                'proxies': {
                        'http': None,
                        'https': None
                'name': 'global',
                'repo_url': 'http: //wapt.tranquilit.local/wapt'
                'dnsdomain': u'tranquilit.local',
                'proxies': {
                        'http': None,
                        'https': None
                'name': 'wapt-host',
                'repo_url': 'http: //srvwapt.tranquilit.local/wapt-host'
        'common-version': '1.1.1',
        'wapt-exe-version': u'',
        'waptservice_port': 8088,
        'wapt-py-version': '1.1.1'

Wapt database


Test reachability of waptserver.

If waptserver is defined and available, return True, else False

Returns:True if server is defined and actually reachable
Return type:boolean

Wapt user session database


Update configuration parameters to supplied inifilename

write_param(name, value)[source]

Store in local db a key/value pair for later use

class common.WaptBaseDB(dbpath)[source]

Bases: object

curr_db_version = None
db = None
get_param(name, default=None)[source]

Retrieve the value associated with name from database

query(query, args=(), one=False)[source]

execute la requete query sur la db et renvoie un tableau de dictionnaires

set_param(name, value)[source]

Store permanently a (name/value) pair in database, replace existing one


Update local database structure to current version if rules are described in db_upgrades

class common.WaptDB(dbpath)[source]

Bases: common.WaptBaseDB

Class to manage SQLite database with local installation status

add_package(package='', version='', section='', priority='', architecture='', maintainer='', description='', filename='', size='', md5sum='', depends='', conflicts='', sources='', repo_url='', repo='')[source]
add_start_install(package, version, architecture, params_dict={}, explicit_by=None)[source]

Register the start of installation in local db params_dict is the dictionary pf parameters provided on command line with –params

or by the server
explicit_by : username of initiator of the install.
if not None, install is not a dependencie but an explicit manual install
setuppy is the python source code used for install, uninstall or session_setup
code used for uninstall or session_setup must use only wapt self library as package content is not longer available at this step.
Given a list of packages conditions (packagename (optionalcondition))
return a list of dependencies (packages conditions) to install
TODO : choose available dependencies in order to reduce the number of new packages to install
>>> waptdb = WaptDB(':memory:')
>>> office = PackageEntry('office','0')
>>> firefox22 = PackageEntry('firefox','22')
>>> firefox22.depends = 'mymissing,flash'
>>> firefox24 = PackageEntry('firefox','24')
>>> thunderbird = PackageEntry('thunderbird','23')
>>> flash10 = PackageEntry('flash','10')
>>> flash12 = PackageEntry('flash','12')
>>> office.depends='firefox(<24),thunderbird,mymissing'
>>> firefox22.depends='flash(>=10)'
>>> firefox24.depends='flash(>=12)'
>>> waptdb.add_package_entry(office)
>>> waptdb.add_package_entry(firefox22)
>>> waptdb.add_package_entry(firefox24)
>>> waptdb.add_package_entry(flash10)
>>> waptdb.add_package_entry(flash12)
>>> waptdb.add_package_entry(thunderbird)
>>> waptdb.build_depends('office')
([u'flash(>=10)', u'firefox(<24)', u'thunderbird'], [u'mymissing'])
curr_db_version = '20140410'

Initialize current sqlite db with empty table and return structure version


Return a PackageEntry of the local install status for id


Return a dictionary of installed packages : keys=package, values = PackageEntry

installed_matching(package_cond, include_errors=False)[source]

Return True if one properly installed (if include_errors=False) package match the package condition ‘tis-package (>=version)’

Return a list of installed package entries based on search keywords


return a list of all (package,version)

package_entry_from_db(package, version_min='', version_max='')[source]

Return the most recent package entry given its packagename and minimum and maximum version

>>> waptdb = WaptDB(':memory:')
>>> waptdb.add_package_entry(PackageEntry('dummy','1'))
>>> waptdb.add_package_entry(PackageEntry('dummy','2'))
>>> waptdb.add_package_entry(PackageEntry('dummy','3'))
>>> waptdb.package_entry_from_db('dummy')
>>> waptdb.package_entry_from_db('dummy',version_min=2)
>>> waptdb.package_entry_from_db('dummy',version_max=1)

Return an ordered list of available packages entries which match the condition “packagename[([=<>]version)]?” version ascending

Return a list of package entries matching the search words


remove references to repo repo_name >>> waptdb = WaptDB(‘c:/wapt/db/waptdb.sqlite’) >>> waptdb.purge_repo(‘main’)

query_package_entry(query, args=(), one=False)[source]

execute la requete query sur la db et renvoie un tableau de PackageEntry Le matching est fait sur le nom de champs.

Les champs qui ne matchent pas un attribut de PackageEntry
sont également mis en attributs !
>>> waptdb = WaptDB(':memory:')
>>> waptdb.add_package_entry(PackageEntry('toto','0',repo='main'))
>>> waptdb.add_package_entry(PackageEntry('dummy','2',repo='main'))
>>> waptdb.add_package_entry(PackageEntry('dummy','1',repo='main'))
>>> waptdb.query_package_entry("select * from wapt_package where package=?",["dummy"])
[PackageEntry('dummy','2'), PackageEntry('dummy','1')]
>>> waptdb.query_package_entry("select * from wapt_package where package=?",["dummy"],one=True)

Remove status of package installation from localdb

store_setuppy(rowid, setuppy=None, install_params={})[source]

Update status of package installation on localdb

switch_to_explicit_mode(package, user_id)[source]

Set package install mode to manual so that package is not removed when meta packages don’t require it anymore

update_install_status(rowid, install_status, install_output, uninstall_key=None, uninstall_string=None)[source]

Update status of package installation on localdb

update_install_status_pid(pid, install_status='ERROR')[source]

Update status of package installation on localdb

update_repos_list(repos_list, proxies=None, force=False)[source]
update the packages database with Packages files from the url repos_list

removes obsolete records for repositories which are no more referenced repos_list : list of all the repositories objects referenced by the system

as returned by Wapt.repositories
force : update repository even if date of packages index is same as
last retrieved date
return a dictionary of update_db results for each repository name
which has been accessed.
>>> wapt = Wapt(config_filename = 'c:/tranquilit/wapt/tests/wapt-get.ini' )
>>> res = wapt.waptdb.update_repos_list(wapt.repositories)

Return a dictionary of upgradable Package entries

class common.WaptHostRepo(url=None, name='', proxies={'http': None, 'https': None}, timeout=2, dnsdomain=None, hosts=[])[source]

Bases: common.WaptRepo

Dummy http repository for host packages

load_config(config, section=None)[source]

Load waptrepo configuration from inifile section. Use name of repo as section name if section is not provided. Use ‘global’ if no section named section in ini file

update_db(force=False, waptdb=None)[source]

get a list of host packages from remote repo

update_host(host, waptdb, force=False)[source]
Update host package from repo.
Stores last-http-date in database/
returns (host package entry,entry date on server)
>>> repo = WaptHostRepo(name='wapt-host',timeout=4)
>>> print repo.dnsdomain
>>> print repo.repo_url
>>> waptdb = WaptDB(':memory:')
>>> repo.update_host('test-dummy',waptdb)
(None, None)
class common.WaptRepo(url=None, name='', proxies={'http': None, 'https': None}, timeout=2, dnsdomain=None)[source]

Bases: waptpackage.WaptRemoteRepo

Gives access to a remote http repository, with a zipped Packages packages index

>>> repo = WaptRepo(name='main',url='http://wapt/wapt',timeout=4)
>>> packages = repo.packages()
>>> len(packages)
Search the nearest working main WAPT repository given the following priority
  • URL defined in ini file
  • first SRV record in the same network as one of the connected network interface
  • first SRV record with the highest weight
  • wapt CNAME in the local dns domain (https first then http)
preference for SRV records is :
same subnet -> priority asc -> weight desc
>>> repo = WaptRepo(name='main',dnsdomain='',timeout=4,url=None)
>>> repo.repo_url
>>> repo = WaptRepo(name='main',url='http://wapt/wapt',timeout=4)
>>> repo.repo_url
load_config(config, section=None)[source]

Load waptrepo configuration from inifile section. Use name of repo as section name if section is not provided. Use ‘global’ if no section named section in ini file


Repository URL

Fixed url if any, else request DNS with a SRV _wapt._tcp.domain query
or a CNAME by the find_wapt_repo_url method.

The URL is queried once and then cached into a local property.

Returns:url to the repository
Return type:str
>>> repo = WaptRepo(name='main',timeout=4)
>>> print repo.dnsdomain
>>> repo = WaptRepo(name='main',timeout=4)
>>> print repo.dnsdomain
>>> print repo.repo_url

called by wapt when network configuration has changed

update_db(force=False, waptdb=None)[source]
Get Packages from http repo and update local package database
return last-update header
The local status DB is updated. Date of index is stored in params table
for further checks.
  • force (bool) – get index from remote repo even if creation date is not newer than the datetime stored in local status database
  • waptdb (WaptDB) – instance of Wapt status database.

date of Packages index

Return type:


>>> import common
>>> repo = common.WaptRepo('main','http://wapt/wapt')
>>> localdb = common.WaptDB('c:/wapt/db/waptdb.sqlite')
>>> last_update = repo.is_available()
>>> repo.update_db(waptdb=localdb) == last_update
class common.WaptServer(url=None, proxies={'http': None, 'https': None}, timeout=2, dnsdomain=None)[source]

Bases: object

Manage connection to waptserver


Search the WAPT server with dns SRV query

preference for SRV is :
same priority asc -> weight desc
>>> WaptServer(dnsdomain='tranquilit.local',timeout=4,url=None).server_url
>>> WaptServer(url='http://srvwapt:8080',timeout=4).server_url
get(action, auth=None, timeout=None)[source]
load_config(config, section='global')[source]

Load waptserver configuration from inifile

post(action, data=None, files=None, auth=None, timeout=None)[source]

called by wapt when network configuration has changed


Retrieve certificate of https server for further checks


Return fixed url if any, else request DNS

>>> server = WaptServer(timeout=4)
>>> print server.dnsdomain
>>> server = WaptServer(timeout=4)
>>> print server.dnsdomain
>>> print server.server_url
upload_package(package, auth=None, timeout=None)[source]
class common.WaptSessionDB(username='')[source]

Bases: common.WaptBaseDB

add_start_install(package, version, architecture)[source]

Register the start of installation in local db

curr_db_version = '20140410'

Initialize current sqlite db with empty table and return structure version

is_installed(package, version)[source]

Remove status of package installation from localdb >>> wapt = Wapt() >>> wapt.forget_packages(‘tis-7zip’) ???


Remove local user status of packages no more installed

update_install_status(rowid, install_status, install_output)[source]

Update status of package installation on localdb

update_install_status_pid(pid, install_status='ERROR')[source]

Update status of package installation on localdb

common.check_is_member_of(huser, group_name)[source]

check if a user is a member of a group huser : handle pywin32 group_name : group as a string >>> from win32security import LogonUser >>> hUser = win32security.LogonUser (‘technique’,’tranquilit’,’xxxxxxx’,win32security.LOGON32_LOGON_NETWORK,win32security.LOGON32_PROVIDER_DEFAULT) >>> check_is_member_of(hUser,’domain admins’) False

common.check_key_password(key_filename, password='')[source]

Check if provided password is valid to read the PEM private key >>> if not os.path.isfile(‘c:/private/test.pem’): ... create_self_signed_key(‘test’,organization=’Tranquil IT’,locality=u’St Sebastien sur Loire’,commonname=’’,email=’’) >>> check_key_password(‘c:/private/test.pem’,’‘) True >>> check_key_password(‘c:/private/ko.pem’,’‘) False

common.check_user_membership(user_name, password, domain_name, group_name)[source]

check if a user is a member of a group user_name: user as a string password: as a string domain_name : as a string. If empty, check local then domain group_name : group as a string >>> from win32security import LogonUser >>> hUser = win32security.LogonUser (‘technique’,’tranquilit’,’xxxxxxx’,win32security.LOGON32_LOGON_NETWORK,win32security.LOGON32_PROVIDER_DEFAULT) >>> check_is_member_of(hUser,’domain admins’) False

common.create_recursive_zip_signed(zipfn, source_root, target_root=u'', excludes=[u'.svn', u'.git', u'.gitignore', u'*.pyc', u'*.dbg', u'src'])[source]

Create a zip file with filename zipf from source_root directory with target_root as new root. Don’t include file which match excludes file pattern add a file WAPT/manifest.sha1 with sha1 hash of all files add a file WAPT/signature with the bas64 encoded signature of WAPT/manifest.sha1

common.create_self_signed_key(orgname, wapt_base_dir=None, destdir='c:\\private', country='FR', locality=u'', organization=u'', unit='', commonname='', email='')[source]

Creates a self signed key/certificate without password return a dict {‘crt_filename’: ‘c:\private\test.crt’, ‘pem_filename’: ‘c:\private\test.pem’} >>> if os.path.isfile(‘c:/private/test.pem’): ... os.unlink(‘c:/private/test.pem’) >>> create_self_signed_key(‘test’,organization=’Tranquil IT’,locality=u’St Sebastien sur Loire’,commonname=’’,email=’’) {‘crt_filename’: ‘c:\private\test.crt’, ‘pem_filename’: ‘c:\private\test.pem’}


callback to extend handling of json.dumps


Check if the file is encoded in utf8 readable encoding without BOM rewrite the file in place if not compliant.

return localized version of domain admin group (ie “domain admins” or
“administrateurs du domaine” with RID -512)
>>> get_domain_admins_group_name()
u'Domain Admins'
common.get_manifest_data(source_root, target_root=u'', excludes=[u'.svn', u'.git', u'.gitignore', u'*.pyc', u'*.dbg', u'src'])[source]

Return a list of [filenames,sha1 hash] from files from source_root directory with target_root as new root. Don’t include file which match excludes file pattern


Retrieve certificate for further checks


return a list of (iface,mac,{addr,broadcast,netmask})

common.html_table(cur, callback=None)[source]

cur est un cursor issu d’une requete callback est une fonction qui prend (rowmap,fieldname,value) et renvoie une representation texte

common.import_code(code, name='', add_to_sys_modules=0)[source]

Import dynamically generated code as a module. code is the object containing the code (a string, a file handle or an actual compiled code object, same types as accepted by an exec statement). The name is the name to give to the module, and the final argument says wheter to add it to sys.modules or not. If it is added, a subsequent import statement using name will return this module. If it is not added to sys.modules import will try to load it in the normal fashion.

import foo

is equivalent to

foofile = open(“/path/to/”) foo = import_code(foofile,”foo”,1)

Returns a newly generated module. From :

common.import_setup(setupfilename, modulename='')[source]

Import setupfilename as modulename, return the module object

common.jsondump(o, **kwargs)[source]

Dump argument to json format, including datetime and customized classes with as_dict or as_json callables >>> class MyClass(object): ... def as_dict(self): ... return {‘test’:’a’,’adate2’,03,15)} >>> jsondump({‘adate’,03,14),’an_object’:MyClass()}) ‘{“adate”: “2014-03-14”, “an_object”: {“test”: “a”, “adate2”: “2014-03-15”}}’

common.lookup_name_from_rid(domain_controller, rid)[source]
return username or group name from RID (with localization if applicable)
from domain_controller : should be a DC rid : integer number (512 for domain admins, 513 for domain users, etc.)
>>> lookup_name_from_rid('srvads', DOMAIN_GROUP_RID_ADMINS)
u'Domain Admins'
common.merge_dict(d1, d2)[source]

merge similar dict

common.ppdicttable(alist, columns=[], callback=None)[source]

pretty print a list of dict as a table columns is an ordered list of (fieldname,width) callback is a function called for each field (fieldname,value) to format the output

common.pptable(cursor, data=None, rowlens=0, callback=None)[source]

pretty print a query result as a table callback is a function called for each field (fieldname,value) to format the output


Return True if key can not be loaded without password >>> private_key_has_password(r’c:/tranquilit/wapt/tests/ssl/test.pem’) False >>> private_key_has_password(r’c:/tmp/ko.pem’) True


Default password callback for opening private keys

common.read_in_chunks(f, chunk_size=131072)[source]

Lazy function (generator) to read a file piece by piece. Default chunk size: 128k.

common.reg_openkey_noredir(key, sub_key, sam=131097)[source]

Return True if computer is connected to AC power supply

common.same_net(ip1, ip2, netmask)[source]

Given 2 ipv4 address and mask, return True if in same subnet

common.sha1_for_file(fname, block_size=1048576)[source]
common.sha256_for_file(fname, block_size=1048576)[source]
common.sid_from_rid(domain_controller, rid)[source]

Return SID structure based on supplied domain controller’s domain and supplied rid rid can be for example DOMAIN_GROUP_RID_ADMINS, DOMAIN_GROUP_RID_USERS

common.splitThousands(s, tSep=', ', dSep='.')[source]

Splits a general float on thousands. GIGO on general input

common.ssl_sign_content(content, private_key, callback=<function pwd_callback at 0x023FD570>)[source]

Sign content with the private_key, return the signature

common.ssl_verify_content(content, signature, public_certs)[source]
Check that the signature matches the content, using the provided list of public keys
Content, signature are String public_certs is either a filename or a list of filenames
>>> if not os.path.isfile('c:/private/test.pem'):
...     key = create_self_signed_key('test',organization='Tranquil IT',locality=u'St Sebastien sur Loire',commonname='',email='')
>>> my_content = 'Un test de contenu'
>>> my_signature = ssl_sign_content(my_content,'c:/private/test.pem')
>>> ssl_verify_content(my_content,my_signature,'c:/private/test.crt')
'C=FR, L=St Sebastien sur Loire, O=Tranquil IT,'
common.tryurl(url, proxies=None, timeout=2, auth=None, verify_cert=False)[source]

Return True if UAC is enabled