Merge pull request #6 from gardenlinux/combine-deb

Add initial tool to combine Debian information

src/glvd/cli/combine_deb.py

@@ -0,0 +1,152 @@
+# SPDX-License-Identifier: MIT
+
+from __future__ import annotations
+
+import logging
+from typing import (
+    AsyncGenerator,
+)
+
+import asyncio
+from sqlalchemy import (
+    bindparam,
+    select,
+    text,
+)
+from sqlalchemy.ext.asyncio import (
+    AsyncEngine,
+    AsyncSession,
+    async_sessionmaker,
+    create_async_engine,
+)
+
+from ..database import Base, DistCpe, DebCve
+
+
+logger = logging.getLogger(__name__)
+
+
+class CombineDeb:
+    stmt_combine_new = (
+        text('''
+            SELECT
+                    cve.cve_id
+                    , src.deb_source
+                    , src.deb_version
+                    , COALESCE(src.deb_version < cve.deb_version_fixed, TRUE) AS debsec_vulnerable
+                FROM
+                    debsrc as src
+                    LEFT OUTER JOIN debsec_cve AS cve ON src.deb_source = cve.deb_source
+                WHERE
+                    src.dist_id = :dist_id
+                    AND cve.dist_id = ANY(:dists_fallback_id)
+        ''')
+        .bindparams(
+            bindparam('dist_id'),
+            bindparam('dists_fallback_id'),
+        )
+    )
+
+    async def combine_dists(
+        self,
+        session: AsyncSession,
+    ) -> AsyncGenerator[tuple[DistCpe, list[DistCpe]], None]:
+        stmt = (
+            select(DistCpe)
+            # Empty version is a fallback, make sure we see them first
+            .order_by(DistCpe.cpe_version)
+        )
+
+        dists_fallback: dict[tuple[str, str], DistCpe] = {}
+
+        async for r in await session.stream(stmt):
+            entry = r[0]
+
+            if entry.cpe_version == '':
+                dists_fallback[(entry.cpe_vendor, entry.cpe_product)] = entry
+            else:
+                dists = [entry]
+                if fallback := dists_fallback.get((entry.cpe_vendor, entry.cpe_product)):
+                    dists.append(fallback)
+                # XXX: Handle fallback to Debian
+                yield entry, dists
+
+    async def combine_update(
+        self,
+        session: AsyncSession,
+        dist: DistCpe,
+        new_entries: dict[tuple[str, str], DebCve],
+    ) -> None:
+        stmt = (
+            select(DebCve)
+            .where(DebCve.dist == dist)
+        )
+
+        async for r in await session.stream(stmt):
+            entry = r[0]
+
+            new_entry = new_entries.pop((entry.cve_id, entry.deb_source), None)
+            if not new_entry:
+                # XXX: Delete entry
+                continue
+
+            # Update object in place. Only real changes will be commited
+            entry.merge(new_entry)
+
+    async def combine_insert(
+        self,
+        session: AsyncSession,
+        dist: DistCpe,
+        new_entries: dict[tuple[str, str], DebCve],
+    ) -> None:
+        for entry in new_entries.values():
+            session.add(entry)
+
+    async def combine(
+        self,
+        session: AsyncSession,
+    ) -> None:
+        # Find all applying CVE for a source.  Was not able to make this work in ORM.
+        async for dist, dists_fallback in self.combine_dists(session):
+            new_entries: dict[tuple[str, str], DebCve] = {}
+
+            async for r in await session.stream(self.stmt_combine_new, {
+                'dist_id': dist.id,
+                'dists_fallback_id': [i.id for i in dists_fallback],
+            }):
+                cve_id, deb_source, deb_version, debsec_vulnerable = r
+
+                new_entries[(cve_id, deb_source)] = DebCve(
+                    dist=dist,
+                    cve_id=cve_id,
+                    deb_source=deb_source,
+                    deb_version=deb_version,
+                    debsec_vulnerable=debsec_vulnerable,
+                )
+
+            await self.combine_update(session, dist, new_entries)
+            await self.combine_insert(session, dist, new_entries)
+
+    async def __call__(
+        self,
+        engine: AsyncEngine,
+    ) -> None:
+        async with engine.begin() as conn:
+            await conn.run_sync(Base.metadata.create_all)
+
+        async with async_sessionmaker(engine)() as session:
+            await self.combine(session)
+            await session.commit()
+
+
+if __name__ == '__main__':
+    import argparse
+    logging.basicConfig(level=logging.DEBUG)
+    parser = argparse.ArgumentParser()
+    args = parser.parse_args()
+    engine = create_async_engine(
+        "postgresql+asyncpg:///",
+        echo=True,
+    )
+    main = CombineDeb()
+    asyncio.run(main(engine))

src/glvd/database/__init__.py

@@ -82,3 +82,20 @@ def merge(self, other: Self) -> None:
         self.deb_version_fixed = other.deb_version_fixed
         self.debsec_tag = other.debsec_tag
         self.debsec_note = other.debsec_note
+
+
+class DebCve(Base):
+    __tablename__ = 'deb_cve'
+
+    dist_id = mapped_column(ForeignKey(DistCpe.id), primary_key=True)
+    cve_id: Mapped[str] = mapped_column(primary_key=True)
+    last_mod: Mapped[datetime] = mapped_column(init=False, server_default=func.now(), onupdate=func.now())
+    deb_source: Mapped[str] = mapped_column(primary_key=True)
+    deb_version: Mapped[str]
+    debsec_vulnerable: Mapped[bool]
+
+    dist: Mapped[Optional[DistCpe]] = relationship(lazy='selectin', default=None)
+
+    def merge(self, other: Self) -> None:
+        self.deb_version = other.deb_version
+        self.debsec_vulnerable = other.debsec_vulnerable