Merge ~litios/ubuntu-cve-tracker:feature/data-lib into ubuntu-cve-tracker:master

[David Fernandez Gonzalez]

This library provides a project-agnostic interface for loading CVEs and Packages.

This library loads all the information from the different sources into well-defined objects with shared interfaces.

The storage classes load handle these objects and handle them to avoid memory duplication.

-----

[David Fernandez Gonzalez (litios)]

Testing script:

```
from datalib import *
import time

now = time.time()

cve_storage = UCTCVEStorage()
cve_storage.load()

package_storage = UCTPackageStorage()
package_storage.load(filter_releases=['bionic', 'esm-apps/bionic', 'esm-infra/bionic'])

usn_storage = USNStorage()
usn_storage.load()

cve_storage.link_pkg_storage(package_storage)
usn_storage.link_pkg_storage(package_storage)
usn_storage.link_cve_storage(cve_storage)

later = time.time()
print('Total time: ', later - now)

gpac = package_storage.get_package('python-idna')
for version in gpac.source_versions:
    print(gpac, gpac.source_info[version]['release'], version)
    for binary in gpac.binaries[version]:
        print(f' - {binary.name} {binary.version} {binary.arches}')

print()
print('------------------')
print('Showing CVE-2024-3651')
print('------------------')
cve = cve_storage.get_cve('CVE-2024-3651')
print(cve.id)
print(cve.description)
for entry in cve.pkg_entries:
    print(' -', entry.cve, entry.pkg, entry.release, entry.status, entry.note)

print()
print('------------------')
print('Loading jammy into the storage')
package_storage.load_release('jammy')
cve = cve_storage.get_cve('CVE-2024-3651')
print('------------------')
print('Showing CVE entries now')
for entry in cve.pkg_entries:
    print(' -', entry.cve, entry.pkg, entry.release, entry.status, entry.note)

print()
print('------------------')
print('Showing python-idna package')
print('------------------')
gpac = package_storage.get_package('python-idna')
for version in gpac.source_versions:
    print(gpac, gpac.source_info[version]['release'], version)
    for binary in gpac.binaries[version]:
        print(f' - {binary.name} {binary.version} {binary.arches}')

print()
print('------------------')
print('Showing USNs for this CVE')
print('------------------')
usns = usn_storage.get_usns_by_cve(cve.id)
for usn in usns:
    print('USN',usn.id)
    print(usn.description)
    for cve in usn.cves:
        print(cve.id)
    print('Affected releases:', ','.join(list(usn.releases.keys())))
    for source_pkg in usn.package_fixed_versions:
        print(source_pkg.name)
        for fixed_version, release in usn.package_fixed_versions[source_pkg]:
            print(' - Fixed at', fixed_version, 'in', release)

print()
print('------------------')
print('Unloading jammy from the storage')
package_storage.unload_release('jammy')

print()
print('------------------')
print('Showing python-idna package')
print('------------------')
gpac = package_storage.get_package('python-idna')
for version in gpac.source_versions:
    print(gpac, gpac.source_info[version]['release'], version)
    for binary in gpac.binaries[version]:
        print(f' - {binary.name} {binary.version} {binary.arches}')

cve = cve_storage.get_cve('CVE-2024-3651')

print()
print('------------------')
print('Showing CVE-2024-3651')
print('------------------')
for entry in cve.pkg_entries:
    print(' -', entry.cve, entry.pkg, entry.release, entry.status, entry.note)

print()
print('------------------')
print('Showing USNs for this CVE')
print('------------------')
usns = usn_storage.get_usns_by_cve(cve.id)
for usn in usns:
    print('USN',usn.id)
    p...

558d0f0... by David Fernandez Gonzalez on 2024-06-04

Revert "[Data lib] Add docs"

This reverts commit eaae95160bbf4c29a076db144afad562c85844e5.

[David Fernandez Gonzalez (litios)]

Docs were not meant to be merged, it was for the PR itself.

Reverting the change so you can still see them but I'll simply erase the 2 last commits when merging.

[Nick Galanis (nickgalanis)]

LGTM! Thanks David, amazing work. It's a very clean and abstract representation of our data. I ran the script you provided and was functional. Some things to consider (as already discussed):
 - Ensure python3.8 compatibility for focal machines
 - Possibly revisit the idea of multi-threading that could cause low-memory machines to crash

We would need to add/change some fields/functions as we move forward, but for now, I think it is safe to merge.

review: Approve

[Diogo Sousa (0xdsousa)]

Absolutely loving the type hints on everything, great work.

9531bb9... by David Fernandez Gonzalez on 2024-06-06

[DATA LIB] Allow getting source versions per release

Signed-off-by: David Fernandez Gonzalez <email address hidden>

6f1f847... by David Fernandez Gonzalez on 2024-06-06

[DATA LIB] Don't link to object if it hasn't been loaded

Signed-off-by: David Fernandez Gonzalez <email address hidden>

[David Fernandez Gonzalez (litios)]

Adding two more commits:

* Include a feature to retrieve source versions from a package based on the release. Test from the example above:

Showing bionic versions {2.6-1, 2.5-1}
Showing bionic versions, including esm releases {2.6-1, 2.6-1ubuntu0.1~esm1, 2.5-1}

* Fix bug for USNs in case the CVE or the package was not loaded when loading the USN. In that case, those won't get added.

296894d... by David Fernandez Gonzalez on 2024-06-07

[Data lib] Use apt_pkg for version comparison

Signed-off-by: David Fernandez Gonzalez <email address hidden>

38ffc5e... by David Fernandez Gonzalez on 2024-06-10

[DATA LIB] Store CVEs, packages and SNs ordered.

For CVE and SNs, sort them based on the integer
values rather than standard ascii comparison.

Signed-off-by: David Fernandez Gonzalez <email address hidden>

[Eduardo Barretto (ebarretto)]

lgtm, thanks!

review: Approve

diff --git a/scripts/datalib/__init__.py b/scripts/datalib/__init__.py
new file mode 100644
index 0000000..24bda8c
--- /dev/null
+++ b/scripts/datalib/__init__.py
@@ -0,0 +1,5 @@
+from .config import *
+from .models import *
+from .uct_models import *
+from .storage import *
+from .uct_storage import *
diff --git a/scripts/datalib/config.py b/scripts/datalib/config.py
new file mode 100644
index 0000000..6869d92
--- /dev/null
+++ b/scripts/datalib/config.py
@@ -0,0 +1,18 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+# Module containing classes that represent CVE and Package data 
+#
+# Author: David Fernandez Gonzalez <[e-mail]>
+# Copyright (C) 2024 Canonical Ltd.
+#
+# This script is distributed under the terms and conditions of the GNU General
+# Public License, Version 2 or later. See http://www.gnu.org/copyleft/gpl.html
+# for details.
+#
+import os
+
+VERBOSE = os.environ.get('UCTLIB_VERBOSE', True)
+THREADING = os.environ.get('UCTLIB_THREADING', True)
+UCT = os.environ.get('UCT')
+PKG_CACHE_DIR = os.environ.get('UCTLIB_PKG_CACHE_DIR', '~/.cache/pkg-cache')
+USN_DATABASE = os.environ.get('UCTLIB_USN_DATABASE_DIR', '~/.cache/usn/database.json')
diff --git a/scripts/datalib/models.py b/scripts/datalib/models.py
new file mode 100644
index 0000000..adc3e72
--- /dev/null
+++ b/scripts/datalib/models.py
@@ -0,0 +1,245 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+# Module containing classes that represent CVE and Package data 
+#
+# Author: David Fernandez Gonzalez <[e-mail]>
+# Copyright (C) 2024 Canonical Ltd.
+#
+# This script is distributed under the terms and conditions of the GNU General
+# Public License, Version 2 or later. See http://www.gnu.org/copyleft/gpl.html
+# for details.
+#
+
+import re
+import datetime
+import apt_pkg
+apt_pkg.init_system()
+
+class Version:
+    version: str
+    
+    def __init__(self, version) -> None:
+        self.version = version
+
+    def compare_version(self, other: 'Version') -> int:
+        if not hasattr(other, 'version'): raise Exception("Cannot compare")
+        return apt_pkg.version_compare(self.version, other.version)
+
+    def __eq__(self, other: 'Version') -> bool:
+        if not hasattr(other, 'version'): raise Exception("Cannot compare")
+        return self.compare_version(other) == 0
+
+    def __lt__(self, other: 'Version') -> bool:
+        if not hasattr(other, 'version'): raise Exception("Cannot compare")
+        return self.compare_version(other) == -1
+
+    def __hash__(self) -> int:
+        return hash(self.version)
+
+    def __repr__(self) -> str:
+        return self.version
+
+class BinaryPackage:
+    name: str
+    version: Version
+    arches: list[str]
+    source: 'SourcePackage'
+    
+    def __init__(self, name: str, version: Version, arches: list[str], source: 'SourcePackage') -> None:
+        self.name = name
+        self.version = version
+        self.arches = arches
+        self.source = source
+    
+    def __eq__(self, other: object) -> bool:
+        return self.name == other.name and \
+            self.version == other.version and \
+            self.source == other.source
+
+class SourcePackage:
+    name: str
+    source_versions: set[Version]
+    binaries: dict[Version:list[BinaryPackage]]
+
+    def __init__(self, name: str, binaries: dict[Version:list[BinaryPackage]]) -> None:
+        self.name = name
+        self.binaries = binaries
+        self.source_versions = set()
+    
+    def __eq__(self, other: object) -> bool:
+        return self.name == other.name and \
+            self.source_versions == other.source_versions and \
+            self.binaries == other.binaries
+    
+    def is_same_package(self, other: object) -> bool:
+        """Is same package but different version"""
+        return self.name == other.name
+
+    def get_source_versions(self) -> set[Version]:
+        return self.source_versions
+
+    def get_binaries(self, version: Version) -> list[BinaryPackage]:
+        if version not in self.binaries: return []
+        return self.binaries[version]
+
+    def add_source_version(self, version: Version) -> None:
+        if version in self.source_versions: return
+        self.source_versions.add(version)
+        self.binaries[version] = []
+
+    def add_binary(self, src_version: Version, binary: BinaryPackage) -> None:
+        if src_version not in self.binaries: return
+        if binary not in self.binaries[src_version]:
+            self.binaries[src_version].append(binary)
+    
+    def is_binary_present(self, src_version: Version, binary: BinaryPackage) -> bool:
+        if src_version not in self.binaries: return False
+        return binary in self.binaries[src_version]
+
+class CVSS:
+    entity: str
+    score: float
+    severity: str
+    vector: str
+
+    def __init__(self, entity: str, score: float, vector: str) -> None:
+        self.entity = entity
+        self.score = score
+        self.vector = vector
+        if self.score == 0.0:
+            self.severity = 'None'
+        elif self.score < 4.0:
+            self.severity = 'Low'
+        elif self.score < 7.0:
+            self.severity = 'Medium'
+        elif self.score < 9.0:
+            self.severity = 'High'
+        elif self.score <= 10.0:
+            self.severity = 'Critical'
+        else:
+            Exception(f'{self.score} is not a valid CVSS score.')
+    
+    def __lt__(self, other):
+        return self.score < other.score
+    
+    def __repr__(self) -> str:
+        return f'[{self.severity}] {self.vector} ({self.score}) -- {self.entity}'
+
+class CVE:
+    id: str
+    ID_REGEX = r'CVE-\d{4}-\d{4,7}'
+    description: str
+    public_date: str
+    cvss: list[CVSS]
+    assigned_to: str
+    notes: list[tuple]
+    references: list[str]
+    priority: str
+    priority_reason: str
+    pkg_entries: list["CVEPkgEntry"]
+
+    def __init__(self, id: str, description: str, public_date: str, cvss: list[CVSS],
+                 assigned_to: str, notes: list[tuple], references: list[str],
+                 priority: str, priority_reason: str) -> None:
+        
+        if not re.search(self.ID_REGEX, id):
+            raise Exception(f'{id} is not a valid CVE ID')
+
+        self.id = id
+        self.description = description.strip()
+        self.public_date = public_date
+        self.cvss = cvss
+        self.assigned_to = assigned_to
+        self.notes = notes
+        self.references = references
+        self.priority = priority
+        self.priority_reason = priority_reason
+        self.pkg_entries = []
+
+    def __lt__(self, other):
+        if not hasattr(other, 'id'): raise Exception("Cannot compare")
+        _, our_year, our_number = self.id.split('-')
+        _, other_year, other_number = other.id.split('-')
+
+        return int(our_year) < int(other_year) or \
+             int(our_year) == int(other_year) and int(our_number) < int(other_number)
+
+    def add_reference(self, ref: str) -> None:
+        if ref not in self.references:
+            self.references.append(ref)
+
+    def add_note(self, author: str, note: str) -> None:
+        self.notes.append((author, note))
+
+    def set_assigned_to(self, assigned_to: str) -> None:
+        self.assigned_to = assigned_to
+
+    def is_package_present(self, package: SourcePackage) -> bool:
+        for cve_entry in self.pkg_entries:
+            if package == cve_entry.pkg:
+                return True
+
+        return False
+
+    def get_package_entry(self, package: SourcePackage) -> 'CVEPkgEntry':
+        for cve_entry in self.pkg_entries:
+            if package == cve_entry.pkg:
+                return cve_entry
+
+        return None
+
+    def add_cve_entry(self, entry: 'CVEPkgEntry') -> bool:
+        for self_entry in self.pkg_entries:
+            if self_entry.pkg == entry.pkg:
+                return False
+        self.pkg_entries.append(entry)
+        return True
+
+    def needs_triage(self) -> bool:
+        if self.priority == 'needs-triage': return True
+        for cve_entry in self.pkg_entries:
+            if cve_entry.status == 'needs-triage':
+                return True
+
+        return False
+
+    def __eq__(self, other: object) -> bool:
+        return self.id == other.id
+
+    def __repr__(self) -> str:
+        return str(vars(self))
+
+
+class CVEPkgEntry:
+    pkg: SourcePackage
+    cve: CVE
+    status: str
+    note: str
+
+    STATUSES = ['needs-triage', 'needed', 'not-affected', 'released', 'ignored', 'deferred', 'DNE', 'pending']
+    def __init__(self, pkg: SourcePackage, cve: CVE, status: str, note: str) -> None:
+        self.pkg = pkg
+        self.cve = cve
+
+        if status not in self.STATUSES:
+            raise Exception(f'{status} is not a valid status')
+
+        self.status = status
+        self.note = note
+
+class SN:
+    id: str
+    cves: set[CVE]
+    packages: set[SourcePackage]
+    package_fixed_versions: dict[SourcePackage:Version]
+    lp_bugs: set[str]
+    description: str
+    date: datetime.date
+
+    def __init__(self, description: str, date: datetime.date, pkg_fixed_info: dict) -> None:
+        self.cves = set()
+        self.packages = set()
+        self.lp_bugs = set()
+        self.description = description
+        self.date = date
+        self.package_fixed_versions = pkg_fixed_info
\ No newline at end of file
diff --git a/scripts/datalib/storage.py b/scripts/datalib/storage.py
new file mode 100644
index 0000000..fc5c271
--- /dev/null
+++ b/scripts/datalib/storage.py
@@ -0,0 +1,86 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+# Module containing classes that represent CVE and Package data 
+#
+# Author: David Fernandez Gonzalez <[e-mail]>
+# Copyright (C) 2024 Canonical Ltd.
+#
+# This script is distributed under the terms and conditions of the GNU General
+# Public License, Version 2 or later. See http://www.gnu.org/copyleft/gpl.html
+# for details.
+#
+from datalib.uct_models import SourcePackage
+from .models import *
+
+class PackageStorage:
+    """
+    Storage class for handling packages.
+    """
+    packages: dict[str: SourcePackage]
+
+    def __init__(self) -> None:
+        self.packages = {}
+
+    def _order_packages(self) -> None:
+        self.packages = dict(sorted(self.packages.items()))
+
+    def get_package(self, pkg_name: str) -> SourcePackage:
+        if pkg_name not in self.packages: return None
+        return self.packages[pkg_name] 
+
+class CVEStorage:
+    """
+    Storage class for handling CVE packages.
+    """
+    cves: dict[str: CVE]
+    package_storage: PackageStorage
+
+    # TODO: add a changed cache so we can cache CVEs and only update when something got loaded
+
+    def __init__(self) -> None:
+        self.cves = {}
+
+    def link_pkg_storage(self, package_storage: PackageStorage) -> None:
+        self.package_storage = package_storage
+
+    def _order_cves(self) -> None:
+        self.cves = dict(sorted(self.cves.items(),
+                                key=lambda value: int(value[0].split('-')[1]) * 10 ** 8 + int(value[0].split('-')[2])))
+
+    def _init_cve(self, cve: CVE, pkg_data: dict) -> None:
+        """Internal function to add CVE object to the class storage"""
+        self.cves[cve.id] = (cve, pkg_data)
+
+    def load_cve(self, cve_id: str) -> None:
+        """Load a CVE into the storage"""
+        raise NotImplementedError
+
+    def get_cve(self, cve_id: str) -> CVE:
+        """
+        Get a CVE from the storage.
+        The package links will be computed when the
+        CVE is requested, extracting them from the internal
+        PackageStorage.
+        """
+        raise NotImplementedError
+
+class SNStorage:
+    sns: dict[str: SN]
+    package_storage: PackageStorage
+    cve_storage: CVEStorage
+
+    def __init__(self) -> None:
+        self.sns = {}
+
+    def _order_sns(self) -> None:
+        self.sns = dict(sorted(self.sns.items(),
+                               key=lambda value: int(value[0].split('-')[0]) * 10 ** 8 + int(value[0].split('-')[1])))
+
+    def link_pkg_storage(self, package_storage: PackageStorage) -> None:
+        self.package_storage = package_storage
+
+    def link_cve_storage(self, cve_storage: CVEStorage) -> None:
+        self.cve_storage = cve_storage
+
+    def load_sn(self, object: SN) -> None:
+        return NotImplementedError
\ No newline at end of file
diff --git a/scripts/datalib/uct_models.py b/scripts/datalib/uct_models.py
new file mode 100644
index 0000000..7b9fe01
--- /dev/null
+++ b/scripts/datalib/uct_models.py
@@ -0,0 +1,259 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+# Module containing classes that represent CVE and Package data 
+#
+# Author: David Fernandez Gonzalez <[e-mail]>
+# Copyright (C) 2024 Canonical Ltd.
+#
+# This script is distributed under the terms and conditions of the GNU General
+# Public License, Version 2 or later. See http://www.gnu.org/copyleft/gpl.html
+# for details.
+#
+
+from datalib.models import CVSS
+from .models import *
+from cve_lib import get_subproject_details, get_active_esm_releases, is_active_release
+
+class Release:
+    name: str
+    version: str
+    codename: str
+    product: str
+    series: str
+    is_esm: bool
+    is_active: bool
+    requires_oval: bool
+    parent: "Release"
+
+    def __init__(self, name: str) -> None:
+        canon, product, series, details = get_subproject_details(name)
+        if not canon:
+            self.name = name
+            self.parent = None
+            return
+        self.name = canon
+        self.product = product
+        self.series = series
+        self.is_esm = self.name in get_active_esm_releases()
+        self.is_active = is_active_release(self.name)
+        self.parent = None
+        self.version = None
+        self.codename = None
+        if not details: return
+        if 'oval' in details:
+            self.requires_oval = details['oval']
+        else:
+            self.requires_oval = True
+        if 'parent' in details:
+            self.parent = Release(details['parent'])
+        if 'version' in details:
+            self.version = details['version']
+        if 'codename' in details:
+            self.codename = details['codename']
+
+    def __eq__(self, value: object) -> bool:
+        return self.name == value.name
+
+    def __repr__(self) -> str:
+        return self.name
+
+    def __hash__(self):
+        return hash(self.name)
+
+    def get_esm_release(self, component: str) -> "Release":
+        esm_releases = get_active_esm_releases()
+        potential_releases = list(filter(lambda release: self.series in release, esm_releases))
+        if len(potential_releases) == 1:
+            return Release(potential_releases[0])
+        elif len(potential_releases) == 2:
+            if component in ['main', 'restricted']:
+                target = 'esm-infra'
+            elif component in ['universe', 'multiverse']:
+                target = 'esm-apps'
+            
+            if target in potential_releases[0]: return Release(potential_releases[0])
+            else: return Release(potential_releases[1])
+
+        return None
+    def is_parent_release(self, release: "Release") -> bool:
+        return release in self.get_all_parents()
+    
+    def get_direct_parent(self) -> "Release":
+        return self.parent
+
+    def get_all_parents(self) -> "Release":
+        release = self
+        all_parents = []
+        while release.parent:
+            release = release.parent
+            all_parents.append(release)
+        return all_parents
+
+    def get_oldest_parent(self) -> "Release":
+        if not self.parent: return None
+        return  self.get_all_parents()[-1]
+
+class UbuntuVersion(Version):
+    def is_esm(self) -> bool:
+        return 'esm' in self.version
+
+class UbuntuBinaryPackage(BinaryPackage):
+    component: str
+
+    def __init__(self, name: str, version: UbuntuVersion, arches: list[str], source: 'UbuntuSourcePackage', component: str) -> None:
+        self.component = component
+        super().__init__(name, version, arches, source)
+    
+class UbuntuSourcePackage(SourcePackage):
+    source_info: dict[UbuntuVersion: dict]
+    binaries: dict[UbuntuVersion:list[UbuntuBinaryPackage]]
+
+    def __init__(self, name, binaries) -> None:
+        self.source_info = {}
+        super().__init__(name, binaries)
+
+    def add_source_version(self, version: UbuntuVersion, release: Release, component: str, pocket: str):
+        if version in self.source_versions: return
+        self.source_info[version] = {
+            'pocket': pocket,
+            'component': component,
+            'release': release
+        }
+        return super().add_source_version(version)
+
+    def get_version_release(self, version: UbuntuVersion) -> Release:
+        if version not in self.source_versions: return None
+        return self.source_info[version]['release']
+
+    def get_version_component(self, version: UbuntuVersion) -> str:
+        if version not in self.source_versions: return None
+        return self.source_info[version]['component']
+    
+    def get_version_pocket(self, version: UbuntuVersion) -> str:
+        if version not in self.source_versions: return None
+        return self.source_info[version]['pocket']
+
+    def get_release_source_versions(self, release: Release, include_parents: bool = False, include_esm: bool = False) -> set[UbuntuVersion]:
+        rel_versions = set()
+        for source_version in self.source_versions:
+            source_version_rel = self.get_version_release(source_version)
+            component = self.get_version_component(source_version)
+            if (include_parents and release.is_parent_release(source_version_rel)) or \
+                (include_esm and source_version_rel == release.get_esm_release(component)) or \
+                source_version_rel == release:
+                rel_versions.add(source_version)
+
+        return rel_versions
+
+    def get_latest_version(self, release: Release = None, include_parents: bool = False, include_esm: bool = False) -> UbuntuVersion:
+        if release:
+            source_versions = self.get_release_source_versions(release, include_parents, include_esm)
+        else:
+            source_versions = self.get_source_versions()
+
+        if not source_versions: return None
+
+        latest = None
+        for source_version in source_versions:
+            if not latest or source_version > latest:
+                latest = source_version
+
+        return latest
+
+    def get_earliest_version(self, release: Release = None, include_parents: bool = False, include_esm: bool = False) -> UbuntuVersion:
+        if release:
+            source_versions = self.get_release_source_versions(release, include_parents, include_esm)
+        else:
+            source_versions = self.get_source_versions()
+
+        if not source_versions: return None
+
+        earliest = None
+        for source_version in source_versions:
+            if not earliest or source_version < earliest:
+                earliest = source_version
+
+        return earliest
+
+    def expand_versions(self, other: "UbuntuSourcePackage") -> None:
+        self.source_versions.update(other.source_versions)
+        self.source_info.update(other.source_info)
+        self.binaries.update(other.binaries)
+    
+    def remove_version(self, version: UbuntuVersion) -> None:
+        self.source_versions.remove(version)
+        del self.source_info[version]
+        del self.binaries[version]
+
+    def release_exists(self, release: Release, include_parents: bool) -> bool:
+        """
+        Check if release exists in the package.
+
+        If include_parents, it will also check if any of the release
+        parents are listed.
+        """
+        for _, entry in self.source_info.items():
+            if release == entry['release']:
+                return True
+            if include_parents and release.is_parent_release(entry['release']): 
+                return True
+        
+        return False
+
+    def __repr__(self) -> str:
+        return f'{self.name}'
+    
+    def __hash__(self) -> int:
+        return hash(self.name)
+
+class UbuntuCVEPkgEntry(CVEPkgEntry):
+    release: Release
+
+    def __init__(self, pkg: UbuntuSourcePackage, cve: 'UbuntuCVE', status: str, note: str, release: Release) -> None:
+        self.release = release
+        super().__init__(pkg, cve, status, note)
+
+class UbuntuCVE(CVE):
+    pkg_entries: list[UbuntuCVEPkgEntry]
+
+    def __init__(self, id: str, description: str, public_date: str, cvss: list[CVSS], assigned_to: str, notes: list[tuple], references: list[str], priority: str, priority_reason: str) -> None:
+        super().__init__(id, description, public_date, cvss, assigned_to, notes, references, priority, priority_reason)
+
+    def is_rel_present(self, release: Release) -> bool:
+        for entry in self.pkg_entries:
+            if entry.release == release: return True
+        return False
+
+    def add_cve_entry(self, entry: UbuntuCVEPkgEntry) -> bool:
+        for self_entry in self.pkg_entries:
+            if self_entry.pkg == entry.pkg and \
+                self_entry.release == entry.release:
+                return False
+        self.pkg_entries.append(entry)
+        return True
+
+    def __repr__(self) -> str:
+        return self.id
+
+    def __hash__(self) -> int:
+        return hash(self.id)
+    
+
+class USN(SN):
+    package_fixed_versions: dict[SourcePackage:list[tuple[Version, Release]]]
+    releases: set[Release]
+
+    def __init__(self, id: str, data: dict, cve_objs: set, package_fixed_versions: dict, lp_bugs: set):
+        for item in ['description', 'releases', 'title', 'timestamp', 'summary', 'action', 'id', 'isummary']:
+            if item in data:
+                setattr(self, item, data[item])
+            else:
+                setattr(self, item, None)
+
+        self.id = id
+        self.lp_bugs = lp_bugs
+        self.cves = cve_objs
+        self.package_fixed_versions = package_fixed_versions
+
+    def is_rel_present(self, release: Release) -> bool:
+        return release in self.releases
diff --git a/scripts/datalib/uct_storage.py b/scripts/datalib/uct_storage.py
new file mode 100644
index 0000000..f0e6232
--- /dev/null
+++ b/scripts/datalib/uct_storage.py
@@ -0,0 +1,322 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+# Module containing classes that represent CVE and Package data 
+#
+# Author: David Fernandez Gonzalez <[e-mail]>
+# Copyright (C) 2024 Canonical Ltd.
+#
+# This script is distributed under the terms and conditions of the GNU General
+# Public License, Version 2 or later. See http://www.gnu.org/copyleft/gpl.html
+# for details.
+#
+from .storage import *
+from .uct_models import *
+from .config import PKG_CACHE_DIR, USN_DATABASE
+from .utils import get_active_ubuntu_releases
+from .config import UCT
+
+import json
+import os
+import cve_lib
+from multiprocessing import Pool
+
+class UCTPackageStorage(PackageStorage):
+    """
+    Storage class for handling UCT packages.
+    """
+    packages: dict[str: UbuntuSourcePackage]
+    latest_date_created: dict[Release:datetime.datetime]
+    loaded_releases: set[Release]
+
+    def __init__(self) -> None:
+        self.loaded_releases = set()
+        self.latest_date_created = {}
+        super().__init__()
+    
+    def load(self, filter_releases: list = []) -> None:
+        final_releases = []
+        active_ubuntu_releases = get_active_ubuntu_releases()
+        for release in active_ubuntu_releases:
+            if filter_releases and release not in filter_releases: continue
+            final_releases.append(release)
+
+        with Pool(processes=os.cpu_count()) as pool:
+            print(f'Loading {len(final_releases)} releases')
+            package_data = pool.map(self._load_release_from_cache, final_releases)
+
+        for release, release_data, last_date_created in package_data:
+            self._init_release(release, release_data, last_date_created)
+
+        self._order_packages()
+
+    def _load_release_from_cache(self, release: str)  -> dict:
+        data = {}
+
+        try:
+            with open(os.path.join(os.path.expanduser(PKG_CACHE_DIR), release.replace('/', '_') + '-pkg-cache.json')) as f:
+                data = json.load(f)
+        except FileNotFoundError as ex:
+            print(ex)
+            return (release, None, None)
+
+        release_data = {}
+        latest_date_created = None
+        release_obj = Release(release)
+        for src_pkg_name, src_pkg_versions in data.items():
+            if src_pkg_name == 'latest_date_created':
+                latest_date_created = datetime.datetime.fromtimestamp(src_pkg_versions)
+                continue
+            source_package = UbuntuSourcePackage(src_pkg_name, {})
+
+            release_data[src_pkg_name] = source_package
+            for src_pkg_version, src_pkg_version_data in src_pkg_versions.items():
+                src_pkg_version_obj = UbuntuVersion(src_pkg_version)
+                source_package.add_source_version(src_pkg_version_obj, release_obj,
+                                                  src_pkg_version_data['component'], 
+                                                  src_pkg_version_data['pocket'])
+
+                for binary_name, binary_data in src_pkg_version_data['binaries'].items():
+                    source_package.add_binary(src_pkg_version_obj, UbuntuBinaryPackage(binary_name, 
+                                                                  UbuntuVersion(binary_data['version']), 
+                                                                  binary_data['arch'], 
+                                                                  source_package,
+                                                                  binary_data['component']))
+
+        return (release_obj, release_data, latest_date_created)
+    
+    def _init_release(self, release: Release, release_data: dict, latest_date_created: str) -> None:
+        if not release_data: return
+        for package_name in release_data:
+            if package_name in self.packages:
+                self.packages[package_name].expand_versions(release_data[package_name])
+            else:
+                self.packages[package_name] = release_data[package_name]
+        
+        self.latest_date_created[release] = latest_date_created
+        self.loaded_releases.add(release)
+
+    def get_package(self, pkg_name: str) -> UbuntuSourcePackage:
+        return super().get_package(pkg_name)
+
+    def load_release(self, release:str) -> None:
+        if Release(release) in self.loaded_releases: return
+        release_obj, data, date = self._load_release_from_cache(release)
+        self._init_release(release_obj, data, date)
+
+    def unload_release(self, release: str) -> None:
+        release = Release(release)
+        if release not in self.loaded_releases: return
+        self.loaded_releases.remove(release)
+
+        for package_name, package in self.packages.items():
+            original_source_versions = package.source_versions.copy()
+            for version in original_source_versions:
+                ver_rel = package.get_version_release(version)
+                if ver_rel == release: package.remove_version(version)
+
+class UCTCVEStorage(CVEStorage):
+    """
+    Storage class for handling UCT CVE packages.
+    """
+    cves: dict[str: UbuntuCVE]
+    package_storage: UCTPackageStorage
+
+    # TODO: add a changed cache so we can cache CVEs and only update when something got loaded
+
+    def load(self, cves_filter: list = []) -> None:
+        final_cves = []
+        cves_ids = self.get_uct_cve_ids()
+        for cve_id in cves_ids:
+            if cves_filter and cve_id not in cves_filter: continue
+            final_cves.append(cve_id)
+
+        with Pool(processes=os.cpu_count()) as pool:
+            print(f'Loading {len(final_cves)} CVEs')
+            cves = pool.map(self._uct_cve_loader, final_cves)
+
+        for cve, cve_pkg_data in cves:
+            if not cve: continue
+            self._init_cve(cve, cve_pkg_data)
+
+        self._order_cves()
+
+    def _uct_cve_loader(self, cve_id: str) -> UbuntuCVE:
+        """Internal function to load CVE from UCT"""
+        data = {}
+        for directory in cve_lib.cve_dirs:
+            try:
+                data = cve_lib.load_cve(os.path.join(UCT, directory, cve_id))
+            except:
+                continue
+        
+        if not data:
+            raise FileNotFoundError(f'Couldn\'t find CVE {cve_id}')
+
+        cvss = []
+        for cvss_entry in data['CVSS']:
+            if not cvss_entry: continue
+
+            cvss.append(CVSS(
+                cvss_entry['source'],
+                float(cvss_entry['baseScore']),
+                cvss_entry['vector']
+            ))
+
+        try:
+            cve = UbuntuCVE(data['Candidate'], data['Description'], data['PublicDate'], cvss, data['Assigned-to'], data['Notes'], data['References'], data['Priority'][0], data['Priority'][1])
+        except Exception as ex:
+            print(ex)
+            return (None, None)
+        
+        return (cve, data['pkgs'])
+
+    def get_uct_cve_ids(self) -> list:
+        """Return all CVE IDs identified from UCT"""
+
+        cve_ids = []
+        for dir in cve_lib.cve_dirs:
+            for (_, _, filenames) in os.walk(dir):
+                for filename in filenames:
+                    if not re.search(CVE.ID_REGEX, filename): continue
+                    cve_ids.append(filename)
+                break
+        
+        return cve_ids
+
+    def load_cve(self, cve_id: str) -> None:
+        """Load a CVE into the storage"""
+        cve, cve_data = self._uct_cve_loader(cve_id)
+        self._init_cve(cve, cve_data)
+
+    def get_all_cves(self) -> list[UbuntuCVE]:
+        """Get all CVEs from the storage"""
+        cves = []
+        for cve_id in self.cves:
+            cves.append(self.get_cve(cve_id))
+        return cves
+
+    def get_all_rel_cves(self, release: Release) -> list[UbuntuCVE]:
+        """Get all CVEs from the storage affecting a release"""
+        cves = []
+        for cve_id in self.cves:
+            cve = self.get_cve(cve_id)
+            if cve.is_rel_present(release):
+                cves.append()
+        return cves
+
+    def populate_cve_entries(self, cve_id: str) -> None:
+        cve, cve_pkg_data = self.cves[cve_id]
+        cve.pkg_entries = list()
+        for package_name, package_entries in cve_pkg_data.items():
+            package = self.package_storage.get_package(package_name)
+            if not package: continue
+            for release, status in package_entries.items():
+                release = Release(release)
+                if not package.release_exists(release, include_parents=True): continue
+                cve.add_cve_entry(UbuntuCVEPkgEntry(package, cve, status[0], status[1], release))
+
+    def get_cve(self, cve_id: str, with_pkg_links: bool = True) -> UbuntuCVE:
+        """
+        Get a CVE from the storage.
+        The package links will be computed when the
+        CVE is requested, extracting them from the internal
+        UCTPackageStorage.
+
+        If with_pkg_links is disabled, links won't be computed
+        """
+        if cve_id not in self.cves: return None
+
+        cve, _ = self.cves[cve_id]
+        cve.pkg_entries = list()
+        if not with_pkg_links: return cve
+        self.populate_cve_entries(cve_id)
+
+        return cve
+
+class USNStorage(SNStorage):
+    sns: dict[str: tuple[USN, list, dict]]
+    package_storage: UCTPackageStorage
+    cve_storage: UCTCVEStorage
+
+    def load(self) -> None:
+        usns_data = self._get_usn_database_data()
+        print(f'Loading {len(usns_data)} USNs')
+        for usn_id, usn_data in usns_data.items():
+            _, _, cves, lp_bugs, pkgs = self.process_usn_data((usn_id, usn_data))
+            self.add_usn(usn_id, usn_data, cves, lp_bugs, pkgs)
+
+        self._order_sns()
+
+    def _get_usn_database_data(self) -> dict:
+        data = {}
+        with open(os.path.expanduser(USN_DATABASE), 'r') as f:
+            data.update(json.load(f))
+        return data
+
+    def process_usn_data(self, usn_raw: dict) -> tuple[set, set, dict]:
+        pkgs = {}
+        lp_bugs = set()
+        cves = set()
+        usn_id = usn_raw[0]
+        usn_data = usn_raw[1]
+        # Package - fixed version loading
+        for rel, info in usn_data['releases'].items():
+            # Get package
+            for pkg, pkg_info in info['sources'].items():
+                pkgs.setdefault(pkg, [])
+                pkgs[pkg].append((pkg_info['version'], rel))
+
+        # CVE loading
+        for cve_text in usn_data['cves']:
+            if 'launchpad.net' in cve_text:
+                lp_bugs.add(cve_text)
+                continue
+            elif re.search(r'CVE-\d{4}-\d{4,7}', cve_text):
+                cves.add(cve_text)
+
+        return usn_id, usn_data, cves, lp_bugs, pkgs 
+        
+    def add_usn(self, usn_id: str, usn_data: dict, cves: set, lp_bugs: set, pkgs: dict) -> None:
+        self.sns[usn_id] = (USN(usn_id, usn_data, set(), dict(), lp_bugs), cves, pkgs)
+
+    def get_usns_by_cve(self, cve_id: str) -> list[USN]:
+        target_usns = []
+        for usn_id, usn_data in self.sns.items():
+            if cve_id in usn_data[1]:
+                target_usns.append(self.get_usn(usn_id))
+        
+        return target_usns
+
+    def get_usn(self, usn_id: str) -> USN:
+        if usn_id not in self.sns: return None
+        usn, cves, pkgs = self.sns[usn_id]
+        cve_mapping = set()
+        pkg_mapping = dict()
+
+        for cve_id in cves:
+            cve = self.cve_storage.get_cve(cve_id)
+            if not cve: continue
+            cve_mapping.add(cve)
+        
+        for pkg_name in pkgs:
+            pkg = self.package_storage.get_package(pkg_name)
+            if not pkg: continue
+            for version, release in pkgs[pkg_name]:
+                release = Release(release)
+                if not pkg.release_exists(release, include_parents=True): continue
+                pkg_mapping.setdefault(pkg, [])
+                version = UbuntuVersion(version)
+                if version.is_esm():
+                    for pkg_version in pkg.source_versions:
+                        if pkg.get_version_release(pkg_version) == release:
+                            release = release.get_esm_release(pkg.get_version_component(pkg_version))
+                            break
+
+                pkg_mapping[pkg].append((version, release))
+
+        usn.cves = cve_mapping
+        usn.package_fixed_versions = pkg_mapping
+        
+        return usn
+
+
diff --git a/scripts/datalib/utils.py b/scripts/datalib/utils.py
new file mode 100644
index 0000000..cc072e0
--- /dev/null
+++ b/scripts/datalib/utils.py
@@ -0,0 +1,20 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+# Module containing classes that represent CVE and Package data 
+#
+# Author: David Fernandez Gonzalez <[e-mail]>
+# Copyright (C) 2024 Canonical Ltd.
+#
+# This script is distributed under the terms and conditions of the GNU General
+# Public License, Version 2 or later. See http://www.gnu.org/copyleft/gpl.html
+# for details.
+#
+import cve_lib
+
+def get_active_ubuntu_releases():
+    active_ubuntu_releases = []
+    for release in cve_lib.all_releases:
+        if release not in cve_lib.eol_releases or release in cve_lib.get_active_releases_with_esm():
+            active_ubuntu_releases.append(release)
+    
+    return active_ubuntu_releases
\ No newline at end of file