chore: upgrade dependencies

component_catalog/models.py

@@ -37,7 +37,6 @@
 from django.utils.text import normalize_newlines
 from django.utils.translation import gettext_lazy as _
 
-from attributecode.model import About
 from cyclonedx import model as cyclonedx_model
 from cyclonedx.model import component as cyclonedx_component
 from cyclonedx.model import contact as cyclonedx_contact
@@ -55,6 +54,7 @@
 from component_catalog.license_expression_dje import get_license_objects
 from component_catalog.license_expression_dje import parse_expression
 from component_catalog.license_expression_dje import render_expression_as_html
+from dejacode_toolkit import aboutcode
 from dejacode_toolkit import download
 from dejacode_toolkit import spdx
 from dejacode_toolkit.purldb import PurlDB
@@ -2335,9 +2335,7 @@ def as_about_yaml(self, extra=None):
         """Return ABOUT file data YAML formatted."""
         about_data = self.as_about(extra)
         if about_data:
-            about = About()
-            about.load_dict(about_data, base_dir="")
-            return about.dumps()
+            return aboutcode.dumps(about_data)
 
     @property
     def about_file_name(self):

dejacode/__init__.py

@@ -96,7 +96,6 @@ def get_git_describe_from_version_file(version_file_location=ROOT_DIR / ".VERSIO
 
 # Turn off the warnings for the following modules.
 warnings.filterwarnings("ignore", module="cyclonedx")
-warnings.filterwarnings("ignore", module="clamd")
 warnings.filterwarnings("ignore", category=FutureWarning, module="rq_scheduler.utils")
 warnings.filterwarnings("ignore", category=FutureWarning, module="django.forms.formsets")
 

dejacode_toolkit/aboutcode.py

@@ -0,0 +1,61 @@
+#
+# Copyright (c) nexB Inc. and others. All rights reserved.
+# DejaCode is a trademark of nexB Inc.
+# SPDX-License-Identifier: AGPL-3.0-only
+# See https://github.com/aboutcode-org/dejacode for support or download.
+# See https://aboutcode.org for more information about AboutCode FOSS projects.
+#
+
+"""
+Serialize data as an ABOUT file.
+https://github.com/aboutcode-org/aboutcode-toolkit/blob/develop/SPECIFICATION.rst
+"""
+
+import saneyaml
+
+# Supported fields, in the order they are written in the ABOUT file
+ABOUT_FIELD_NAMES = [
+    "about_resource",
+    "name",
+    "version",
+    "download_url",
+    "description",
+    "homepage_url",
+    "package_url",
+    "notes",
+    "license_expression",
+    "copyright",
+    "notice_file",
+    "notice_url",
+    "redistribute",
+    "attribute",
+    "track_changes",
+    "owner",
+    "owner_url",
+    "contact",
+    "vcs_repository",
+    "checksum_md5",
+    "checksum_sha1",
+    "licenses",
+]
+
+
+def normalize_text(text):
+    """Return `text` with LF line endings, without trailing spaces and surrounding blank lines."""
+    return "\n".join(line.rstrip() for line in text.splitlines()).strip()
+
+
+def dumps(about_data):
+    """
+    Return the `about_data` dict as an ABOUT file YAML string.
+    Fields are written in the `ABOUT_FIELD_NAMES` order, empty and False values are skipped.
+    """
+    about_fields = {}
+    for field_name in ABOUT_FIELD_NAMES:
+        value = about_data.get(field_name)
+        if isinstance(value, str):
+            value = normalize_text(value)
+        if value:
+            about_fields[field_name] = value
+
+    return saneyaml.dump(about_fields)

dejacode_toolkit/clamav.py

@@ -0,0 +1,98 @@
+#
+# Copyright (c) nexB Inc. and others. All rights reserved.
+# DejaCode is a trademark of nexB Inc.
+# SPDX-License-Identifier: AGPL-3.0-only
+# See https://github.com/aboutcode-org/dejacode for support or download.
+# See https://aboutcode.org for more information about AboutCode FOSS projects.
+#
+
+"""
+Minimal ClamAV client: scan a file-like object with the clamd INSTREAM command.
+Standard library only, so the module can be reused outside of DejaCode.
+"""
+
+import re
+import socket
+import struct
+
+DEFAULT_TCP_PORT = 3310
+DEFAULT_UNIX_SOCKET_PATH = "/var/run/clamav/clamd.ctl"
+
+# Each chunk must be smaller than the StreamMaxLength value of clamd.conf
+CHUNK_SIZE = 64 * 1024
+
+# The "n" prefix makes clamd terminate its reply with a newline
+INSTREAM_COMMAND = b"nINSTREAM\n"
+SIZE_LIMIT_EXCEEDED_RESPONSE = "INSTREAM size limit exceeded. ERROR"
+SCAN_RESPONSE_REGEX = re.compile(r"^stream: (?:(?P<reason>.+) )?(?P<status>OK|FOUND|ERROR)$")
+
+
+class ClamdError(Exception):
+    pass
+
+
+class StreamTooLargeError(ClamdError):
+    pass
+
+
+def open_connection(address, timeout=None):
+    """Return a socket connected to clamd at a Unix socket path or a (host, port) tuple."""
+    if not isinstance(address, str):
+        return socket.create_connection(address, timeout=timeout)
+
+    unix_socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+    unix_socket.settimeout(timeout)
+    try:
+        unix_socket.connect(address)
+    except OSError:
+        unix_socket.close()
+        raise
+    return unix_socket
+
+
+def send_stream(clamd_socket, stream):
+    """Send `stream` as length-prefixed chunks, ended by a zero-length chunk."""
+    clamd_socket.sendall(INSTREAM_COMMAND)
+    while chunk := stream.read(CHUNK_SIZE):
+        clamd_socket.sendall(struct.pack("!L", len(chunk)) + chunk)
+    clamd_socket.sendall(struct.pack("!L", 0))
+
+
+def read_response(clamd_socket):
+    with clamd_socket.makefile("rb") as response_file:
+        return response_file.readline().decode("utf-8").strip()
+
+
+def parse_response(response):
+    """Return a (status, reason) tuple: ("OK", None), ("FOUND", signature), ("ERROR", reason)."""
+    if response == SIZE_LIMIT_EXCEEDED_RESPONSE:
+        raise StreamTooLargeError(response)
+
+    response_match = SCAN_RESPONSE_REGEX.match(response)
+    if not response_match:
+        raise ClamdError(f"Unexpected clamd response: {response!r}")
+
+    return response_match.group("status", "reason")
+
+
+def scan_stream(stream, address, timeout=None):
+    """
+    Scan the `stream` file-like object with the clamd service at `address`.
+    Return a (status, reason) tuple, see `parse_response`.
+    """
+    try:
+        clamd_socket = open_connection(address, timeout)
+    except OSError as error:
+        raise ClamdError(f"Cannot connect to clamd at {address}: {error}") from error
+
+    with clamd_socket:
+        try:
+            send_stream(clamd_socket, stream)
+            response = read_response(clamd_socket)
+        except BrokenPipeError as error:
+            # clamd closes the connection once the stream exceeds StreamMaxLength
+            raise StreamTooLargeError("Stream exceeds the clamd StreamMaxLength") from error
+        except OSError as error:
+            raise ClamdError(f"Communication error with clamd at {address}: {error}") from error
+
+    return parse_response(response)

dejacode_toolkit/tests/test_aboutcode.py

@@ -0,0 +1,73 @@
+#
+# Copyright (c) nexB Inc. and others. All rights reserved.
+# DejaCode is a trademark of nexB Inc.
+# SPDX-License-Identifier: AGPL-3.0-only
+# See https://github.com/aboutcode-org/dejacode for support or download.
+# See https://aboutcode.org for more information about AboutCode FOSS projects.
+#
+
+from unittest import TestCase
+
+from dejacode_toolkit import aboutcode
+
+
+class AboutCodeTestCase(TestCase):
+    def test_aboutcode_dumps_field_order(self):
+        about_data = {
+            "licenses": [{"key": "mit", "name": "MIT License", "file": "mit.LICENSE"}],
+            "checksum_sha1": "sha1",
+            "owner": "Owner",
+            "attribute": True,
+            "license_expression": "mit",
+            "version": "1.0",
+            "name": "name",
+            "about_resource": "package.zip",
+        }
+        expected = (
+            "about_resource: package.zip\n"
+            "name: name\n"
+            "version: '1.0'\n"
+            "license_expression: mit\n"
+            "attribute: yes\n"
+            "owner: Owner\n"
+            "checksum_sha1: sha1\n"
+            "licenses:\n"
+            "  - key: mit\n"
+            "    name: MIT License\n"
+            "    file: mit.LICENSE\n"
+        )
+        self.assertEqual(expected, aboutcode.dumps(about_data))
+
+    def test_aboutcode_dumps_skips_empty_false_and_unsupported_values(self):
+        about_data = {
+            "name": "name",
+            "version": "",
+            "description": "  \n ",
+            "notes": None,
+            "redistribute": False,
+            "track_changes": True,
+            "licenses": [],
+            "unsupported_field": "value",
+        }
+        expected = "name: name\ntrack_changes: yes\n"
+        self.assertEqual(expected, aboutcode.dumps(about_data))
+
+    def test_aboutcode_dumps_normalizes_text(self):
+        about_data = {
+            "name": "  name  ",
+            "copyright": "Copyright A  \r\nCopyright B\r\n",
+        }
+        expected = "name: name\ncopyright: |\n  Copyright A\n  Copyright B\n"
+        self.assertEqual(expected, aboutcode.dumps(about_data))
+
+    def test_aboutcode_normalize_text(self):
+        inputs = [
+            # (text, expected)
+            ("", ""),
+            ("text", "text"),
+            ("  text  ", "text"),
+            ("line1 \r\nline2\t\r\n", "line1\nline2"),
+            ("\n\nline1\n\n  indented\n\n", "line1\n\n  indented"),
+        ]
+        for text, expected in inputs:
+            self.assertEqual(expected, aboutcode.normalize_text(text))