Merge pull request #25 from beebeeep/develop

0.7.17

CHANGELOG.md

@@ -1,3 +1,7 @@
+## [Version 0.7.17](https://github.com/beebeeep/cacus/tree/v0.7.17) (2017-09-04)
+  * Added package purging
+  * Purge old packages
+
 ## [Version 0.7.16](https://github.com/beebeeep/cacus/tree/v0.7.16) (2017-08-15)
   * fixed error handling in /api/v1/package/remove
 

cacus/repo_daemon.py

@@ -415,6 +415,12 @@ def post(self, distro):
         simple = req['simple']
         retention = req.get('retention', 0)
         gpg_key = req.get('gpg_key', None)
+        quota = req.get('quota', None)
+
+        if quota is not None:
+            # quota management is available only for admins
+            yield self._check_token(common.Cacus.admin_access)
+
         if not simple:
             gpg_check = req['gpg_check']
             strict = req['strict']
@@ -424,7 +430,7 @@ def post(self, distro):
 
         try:
             old = yield self.settings['workers'].submit(self.settings['manager'].create_distro, distro=distro, description=description,
-                                                        components=comps, gpg_check=gpg_check, strict=strict, simple=simple,
+                                                        components=comps, gpg_check=gpg_check, strict=strict, simple=simple, quota=quota,
                                                         retention=retention, incoming_wait_timeout=incoming_wait_timeout, gpg_key=gpg_key)
             if not old:
                 self.set_status(201)
@@ -569,6 +575,25 @@ def post(self, distro=None, comp=None):
             self.write({'success': False, 'msg': e.message})
 
 
+class ApiPkgPurgeHandler(ApiRequestHandler):
+
+    @gen.coroutine
+    def post(self, distro=None, comp=None):
+        yield self._check_token(distro)
+        req = self._get_json_request()
+        pkg = req['pkg']
+        ver = req['ver']
+        arch = req.get('arch', None)
+
+        try:
+            r = yield self.settings['workers'].submit(self.settings['manager'].purge_package, distro=distro,
+                                                      pkg=pkg, ver=ver, arch=arch)
+            self.write({'success': True, 'msg': r})
+        except common.CacusError as e:
+            self.set_status(e.http_code)
+            self.write({'success': False, 'msg': e.message})
+
+
 class ApiPkgSearchHandler(ApiRequestHandler):
 
     @gen.coroutine
@@ -661,6 +686,7 @@ def _make_app(config):
     api_pkg_upload_re = s['repo_base'] + r"/api/v1/package/upload/(?P<distro>[-_.A-Za-z0-9]+)/(?P<comp>[-_a-z0-9]+)$"
     api_pkg_copy_re = s['repo_base'] + r"/api/v1/package/copy/(?P<distro>[-_.A-Za-z0-9]+)$"
     api_pkg_remove_re = s['repo_base'] + r"/api/v1/package/remove/(?P<distro>[-_.A-Za-z0-9]+)/(?P<comp>[-_a-z0-9]+)$"
+    api_pkg_purge_re = s['repo_base'] + r"/api/v1/package/purge/(?P<distro>[-_.A-Za-z0-9]+)$"
     api_pkg_search_re = s['repo_base'] + r"/api/v1/package/search(?:/(?P<distro>[-_.A-Za-z0-9]+))?$"
     # Distribution operations
     api_distro_create_re = s['repo_base'] + r"/api/v1/distro/create/(?P<distro>[-_.A-Za-z0-9]+)$"
@@ -679,6 +705,7 @@ def _make_app(config):
         url(api_pkg_upload_re, ApiPkgUploadHandler),
         url(api_pkg_copy_re, ApiPkgCopyHandler),
         url(api_pkg_remove_re, ApiPkgRemoveHandler),
+        url(api_pkg_purge_re, ApiPkgPurgeHandler),
         url(api_pkg_search_re, ApiPkgSearchHandler),
         url(api_distro_create_re, ApiDistroCreateHandler),
         url(api_distro_remove_re, ApiDistroRemoveHandler),

cacus/repo_manage.py

@@ -16,15 +16,21 @@
 
 class RepoManager(common.Cacus):
 
-    def create_distro(self, distro, description, components,  simple, strict=None,
+    def create_distro(self, distro, description, components,  simple, strict=None, quota=None,
                       gpg_check=None, incoming_wait_timeout=None, retention=None, gpg_key=None):
         if gpg_key and not self._check_key(gpg_key):
             raise common.CacusError("Cannot find key {} in keychain".format(gpg_key))
         old_distro = self.db.cacus.distros.find_one_and_update(
             {'distro': distro},
-            {'$set': {
-                'gpg_check': gpg_check, 'strict': strict, 'simple': simple, 'retention': retention,
-                'description': description, 'incoming_wait_timeout': incoming_wait_timeout, 'gpg_key': gpg_key}},
+            {
+                '$set': {
+                    'gpg_check': gpg_check, 'strict': strict, 'simple': simple, 'retention': retention, 'quota': quota,
+                    'description': description, 'incoming_wait_timeout': incoming_wait_timeout, 'gpg_key': gpg_key
+                },
+                '$inc': {
+                    'quota_used': 0
+                }
+            },
             return_document=ReturnDocument.BEFORE,
             upsert=True)
         self.log.info("%s distro '%s', simple: %s, components: %s", "Updated" if old_distro else "Created", distro, simple, ', '.join(components))
@@ -135,11 +141,11 @@ def _apply_retention_policy(self, distro, comp, sources, debs, skipUpdateMeta=Fa
         for source in sources_to_delete:
             self.log.warn("Removing %s_%s from %s/%s due to retention policy", source['Package'], source['Version'], distro, comp)
             self.remove_package(pkg=source['Package'], ver=source['Version'], distro=distro, comp=comp,
-                                source_pkg=True, skipUpdateMeta=skipUpdateMeta, locked=True)
+                                source_pkg=True, skipUpdateMeta=skipUpdateMeta, locked=True, purge=True)
         for deb in debs_to_delete:
             self.log.warn("Removing %s_%s_%s from %s/%s due to retention policy", deb['Package'], deb['Version'], deb['Architecture'], distro, comp)
             self.remove_package(pkg=deb['Package'], ver=deb['Version'], distro=distro, comp=comp,
-                                source_pkg=False, skipUpdateMeta=skipUpdateMeta, locked=True)
+                                source_pkg=False, skipUpdateMeta=skipUpdateMeta, locked=True, purge=True)
 
     def _process_deb_file(self, file):
         with open(file) as f:
@@ -255,6 +261,18 @@ def upload_package(self, distro, comp, files, changes, skipUpdateMeta=False):
             skipUpdateMeta - whether to update distro metadata
         """
 
+        distro_settings = self.db.cacus.distros.find_one({'distro': distro}, {'distro': 1, 'strict': 1, 'quota': 1, 'quota_used': 1})
+        incoming_bytes = 0
+        if not distro_settings:
+            raise common.NotFound("Distribution '{}' was not found".format(distro))
+
+        if distro_settings['strict'] and not any(x.endswith('.changes') for x in files):
+            raise common.FatalError("Strict mode enabled for '{}', will not upload package without signed .changes file".format(distro))
+        if distro_settings['quota'] is not None:
+            incoming_bytes = sum(os.stat(file).st_size for file in files)
+            if incoming_bytes > distro_settings['quota'] - distro_settings['quota_used']:
+                raise common.FatalError("Quota exceeded for distro '{distro}': you are using {quota_used} bytes of {quota}".format(**distro_settings))
+
         src_pkg = {}
         debs = []
         affected_arches = set()
@@ -318,6 +336,14 @@ def upload_package(self, distro, comp, files, changes, skipUpdateMeta=False):
                             upsert=True)
                         components_to_update.update(result['components'])
 
+                    if distro_settings['quota'] is not None:
+                        distro_settings = self.db.cacus.distros.find_one_and_update(
+                            {'distro': distro},
+                            {'$inc': {'quota_used': incoming_bytes}},
+                            return_document=ReturnDocument.AFTER,
+                            upsert=False)
+                        self.log.info("Updated quotas for distro %s: used %s out of %s", distro, distro_settings['quota_used'], distro_settings['quota'])
+
                     self._apply_retention_policy(distro, comp, sources=[src_pkg], debs=debs, skipUpdateMeta=skipUpdateMeta)
 
                     if not skipUpdateMeta:
@@ -497,7 +523,11 @@ def _generate_packages_file(self, distro, comp, arch):
         data.write("\n")
         return data
 
-    def remove_package(self, pkg=None,  ver=None, arch=None, distro=None, comp=None, source_pkg=False, skipUpdateMeta=False, locked=False):
+    def remove_package(self, pkg=None,  ver=None, arch=None, distro=None, comp=None, source_pkg=False, purge=False, skipUpdateMeta=False, locked=False):
+        """ Removes package from specified distro/component """
+
+        self.log.info("Removing %s_%s from %s/%s", pkg, ver, distro, comp)
+
         affected_arches = []
         try:
             with common.DistroLock(self.db, distro, [comp], already_locked=locked):
@@ -506,9 +536,8 @@ def remove_package(self, pkg=None,  ver=None, arch=None, distro=None, comp=None,
                     result = self.db.sources[distro].find_one_and_update(
                         {'Package': pkg, 'Version': ver, 'components': comp},
                         {'$pullAll': {'components': [comp]}},
-                        projection={'Architecture': 1},
                         upsert=False,
-                        return_document=ReturnDocument.BEFORE
+                        return_document=ReturnDocument.AFTER
                     )
                     if result:
                         binaries = self.db.packages[distro].update_many(
@@ -524,27 +553,78 @@ def remove_package(self, pkg=None,  ver=None, arch=None, distro=None, comp=None,
                     result = self.db.packages[distro].find_one_and_update(
                         {'Package': pkg, 'Version': ver, 'Architecture': arch, 'components': comp},
                         {'$pullAll': {'components': [comp]}},
-                        projection={'Architecture': 1},
                         upsert=False,
-                        return_document=ReturnDocument.BEFORE
+                        return_document=ReturnDocument.AFTER
                     )
                     affected_arches = [result['Architecture']] if result else []
                 if not result:
                     msg = "Cannot find package '{}_{}' in '{}/{}'".format(pkg, ver, distro, comp)
                     self.log.error(msg)
                     raise common.NotFound(msg)
+
+                msg = "Package '{}_{}' was removed from '{}/{}'".format(pkg, ver, distro, comp)
+                self.log.info(msg)
+
+                if purge and result['components'] == []:
+                    self.purge_package(pkg=result['Package'], ver=result['Version'], distro=distro, skipUpdateMeta=True, locked=True)
+
+                if not skipUpdateMeta:
+                    if 'all' in affected_arches:
+                        affected_arches = None      # update all arches in case we have 'all' arch in scope
+                    self.log.info("Updating '%s' distro metadata for component %s, arch: %s", distro, comp, affected_arches)
+                    self.update_distro_metadata(distro, [comp], affected_arches)
+                return msg
+        except common.DistroLockTimeout as e:
+            raise common.TemporaryError(e.message)
+
+    def purge_package(self, pkg=None,  ver=None, arch=None, distro=None, skipUpdateMeta=False, locked=False):
+        """ Removes package from all components and wipes it (all sources, debs etc) from storage
+        XXX: note that purging package may break existing distro snapshots!
+        """
+
+        self.log.info("Purging %s_%s from distro %s", pkg, ver, distro)
+
+        affected_arches = set()
+        affected_comps = set()
+        try:
+            with common.DistroLock(self.db, distro, comps=None, already_locked=locked):
+                result = self.db.sources[distro].find_one_and_delete({'Package': pkg, 'Version': ver})
+                if result:
+                    # found source package matching query, remove this package, its non-deb files and all debs it consists of
+                    for f in result['files']:
+                        self.storage.delete(f['storage_key'])
+                    source = result['_id']
+                    while True:
+                        result = self.db.packages[distro].find_one_and_delete({'source': source})
+                        if result:
+                            affected_arches.add(result['Architecture'])
+                            affected_comps.update(result['components'])
+                            self.storage.delete(result['storage_key'])
+                        else:
+                            break
                 else:
-                    msg = "Package '{}_{}' was removed from '{}/{}'".format(pkg, ver, distro, comp)
-                    self.log.info(msg)
-                    if not skipUpdateMeta:
-                        if 'all' in affected_arches:
-                            affected_arches = None      # update all arches in case we have 'all' arch in scope
-                        self.log.info("Updating '%s' distro metadata for component %s, arch: %s", distro, comp, affected_arches)
-                        self.update_distro_metadata(distro, [comp], affected_arches)
-                    return msg
+                    # try to find in packages db
+                    selector = {'Package': pkg, 'Version': ver}
+                    if arch:
+                        selector['Architecture'] = arch
+
+                    result = self.db.packages[distro].find_one_and_delete(selector)
+                    if result:
+                        affected_arches.update(result['Architecture'])
+                        affected_comps.update(result['components'])
+                        self.storage.delete(result['storage_key'])
+                    else:
+                        raise common.NotFound("Package not found")
+
+                if not skipUpdateMeta:
+                    if 'all' in affected_arches:
+                        affected_arches = None      # update all arches in case we have 'all' arch in scope
+                    self.update_distro_metadata(distro, affected_comps, affected_arches)
         except common.DistroLockTimeout as e:
             raise common.TemporaryError(e.message)
 
+        return "Package {}_{} was removed from {}".format(pkg, ver, distro)
+
     def copy_package(self, pkg=None, ver=None, arch=None, distro=None, src=None, dst=None, source_pkg=False, skipUpdateMeta=False):
         affected_arches = []
         if not self.db.cacus.components.find_one({'distro': distro, 'component': dst}, {'_id': 1}):
@@ -658,7 +738,6 @@ def create_snapshot(self, distro, name, from_snapshot=None, allow_update=True):
         if from_snapshot:
             distro = self._get_snapshot_name(distro, from_snapshot)
 
-
         existing = self.db.cacus.distros.find_one({'distro': snapshot_name})
         if existing:
             if allow_update:

debian/changelog

@@ -1,8 +1,15 @@
+cacus (0.7.17) unstable; urgency=medium
+
+  * Added package purging
+  * Purge old packages
+
+ -- Cacus maintainer entity <[email protected]>  Mon, 04 Sep 2017 11:34:03 +0000
+
 cacus (0.7.16) unstable; urgency=medium
 
   * fixed error handling in /api/v1/package/remove
 
- -- Daniil Migalin <[email protected]>  Tue, 15 Aug 2017 13:29:40 +0000
+ -- Cacus maintainer entity <[email protected]>  Tue, 15 Aug 2017 13:29:40 +0000
 
 cacus (0.7.15) unstable; urgency=medium
 
@@ -15,7 +22,7 @@ cacus (0.7.14) unstable; urgency=medium
   * Fixed distro import
   * Added CORS stuff
 
- -- Daniil Migalin <[email protected]>  Wed, 02 Aug 2017 10:03:49 +0000
+ -- Cacus maintainer entity <[email protected]>  Wed, 02 Aug 2017 10:03:49 +0000
 
 cacus (0.7.13) unstable; urgency=medium
 
@@ -36,7 +43,7 @@ cacus (0.7.11) unstable; urgency=medium
   * Added list of components to /distro/show
   * Removed old API endpoint
 
- -- Daniil Migalin <[email protected]>  Tue, 27 Jun 2017 12:14:00 +0000
+ -- Cacus maintainer entity <[email protected]>  Tue, 27 Jun 2017 12:14:00 +0000
 
 cacus (0.7.10) unstable; urgency=medium
 

setup.py

@@ -4,7 +4,7 @@
 
 setup(
     name="cacus",
-    version="0.7.16",
+    version="0.7.17",
     author="Danila Migalin",
     author_email="[email protected]",
     url="https://github.com/beebeeep/cacus",

swagger/swagger.yaml

@@ -657,6 +657,10 @@ definitions:
         type: integer
         description: Distro retention policy. How many versions of same package will be kept in distro. Note that old packages will be removed only from distro indices, not from storage.
         default: 0
+      quota:
+        type: integer
+        default: null
+        description: Quota for uploaded packages size, in bytes. Setting this variable require token with admin access. Missing value means no quotas. 
       gpg_key:
         type: string
         description: GPG key to use for signing distro Release file. If omitted, default one (from config) will be used.

tests/fixtures/cacus.py

@@ -22,22 +22,25 @@
 
 
 @pytest.fixture
-def distro(repo_manager):
-    repo_manager.create_distro('testdistro', 'description',
-                               components=['comp1', 'comp2'],
-                               gpg_check=False, strict=False, simple=True,
-                               incoming_wait_timeout=10, retention=2)
-    return {'distro': 'testdistro', 'components': ['comp1', 'comp2']}
+def distro_gen(repo_manager):
 
+    class Generator(object):
+        def get(self, name='testdistro', description='desription', components=['comp1', 'comp2'], gpg_check=False,
+                strict=False, simple=True, incoming_wait_timeout=10, retention=2, quota=None):
+            repo_manager.create_distro(name, description, components=components, gpg_check=gpg_check,
+                                       strict=strict, simple=simple, retention=retention,
+                                       incoming_wait_timeout=incoming_wait_timeout, quota=quota)
+            return {'distro': name, 'components': components}
+
+    return Generator()
 
 @pytest.fixture
-def full_distro(repo_manager):
-    repo_manager.create_distro('testdistro_full', 'description',
-                               components=['comp1', 'comp2'],
-                               gpg_check=False, strict=True, simple=False,
-                               incoming_wait_timeout=10)
-    return {'distro': 'testdistro_full', 'components': ['comp1', 'comp2']}
+def distro(distro_gen):
+    return distro_gen.get()
 
+@pytest.fixture
+def full_distro(distro_gen):
+    return distro_gen.get(simple=False, strict=True)
 
 @pytest.yield_fixture(scope='session')
 def storage():
@@ -77,7 +80,9 @@ def cacus_config(request, storage):
 def repo_manager(request, cacus_config, mongo):
 
     repo_manage = importlib.import_module('cacus.repo_manage')
-    return repo_manage.RepoManager(config=cacus_config, mongo=mongo)
+    manager = repo_manage.RepoManager(config=cacus_config, mongo=mongo)
+    manager.common = importlib.import_module('cacus.common')
+    return manager
 
 
 @pytest.yield_fixture
@@ -90,7 +95,12 @@ def duploader(request, cacus_config, mongo):
     os.kill(duploader_process.pid, signal.SIGTERM)
 
 
-def package_is_in_repo(manager, package, distro, component):
+def package_is_in_repo(manager, package, distro, component, meta=True):
+    if meta:
+        metadata = manager.db.packages[distro].find_one({'Package': package['Package'], 'Version': package['Version']})
+        if not metadata:
+            return False
+
     packages = manager.db.cacus.repos.find_one({
         'distro': distro, 'component': component, 'architecture': package['Architecture']})['packages_file']
     with open(os.path.join(manager.config['storage']['path'], packages)) as f:

tests/fixtures/contrib/testpackage/debian/install

@@ -1 +1,2 @@
 hello /usr/bin/
+data /usr/hello/data