Commit 80499802 authored by Pierre Smeyers's avatar Pierre Smeyers
Browse files

fix: manage projects creation

parent 09869273

logo.png

−59.9 KiB (40.9 KiB)
Loading image diff...

sbom_scanner/scan.py

+55 −5
Original line number Diff line number Diff line
@@ -150,7 +150,8 @@ class Scanner: 
            return project_path

        project_def = DtProjectDef(project_path[-1])

        if project_def.is_uuid:

            # TODO: check if project exists

            # TODO: check if project exists?

            print(f"     ... {AnsiColors.YELLOW}{'/'.join(project_path)}{AnsiColors.RESET} is UUID: assume exists")

            return project_path



        # project is defined by name/version...
@@ -160,7 +161,7 @@ class Scanner: 
            params={"name": project_def.name},

        )

        resp.raise_for_status()

        prj = next(

        matching_prj = next(

            filter(

                lambda prj: prj["name"] == project_def.name

                and prj.get("version") == project_def.version,
@@ -168,13 +169,59 @@ class Scanner: 
            ),

            None,

        )

        if prj:

        if matching_prj:

            # project already exists: replace name with found UUID

            project_path[-1] = prj["uuid"]

            print(

                f"     ... {AnsiColors.YELLOW}{'/'.join(project_path)}{AnsiColors.RESET} found (by name/version): {matching_prj['uuid']}"

            )

            project_path[-1] = "#" + matching_prj["uuid"]

            return project_path



        # project does not exist: create it

        # TODO - unfinished

        # TODO: smart classifier

        data = {

            "name": project_def.name,

            "version": project_def.version,

            "classifier": "APPLICATION",

            "active": True,

        }

        # TODO: externalReferences

        # data["externalReferences"] = [{"type":"vcs","url":project_url}],

        if len(project_path) > 1:

            # project to create is not a root project: retrieve parent

            parent_def = DtProjectDef(project_path[-2])

            if not parent_def.is_uuid:

                # create parent project

                resolved_path = self.create_parent_projects(project_path[:-1])

                # now parent def must be a UUID

                parent_def = DtProjectDef(resolved_path[-1])

            # add parent UUID to params

            data["parent"] = {"uuid": parent_def.uuid}



        print(

            f"     ... {AnsiColors.YELLOW}{'/'.join(project_path)}{AnsiColors.RESET} doesn't exist: create with params {AnsiColors.HGRAY}{json.dumps(data)}{AnsiColors.RESET}..."

        )

        resp = requests.put(

            f"{self.base_api_url}/v1/project",

            headers={

                "X-API-Key": self.api_key,

                "accept": "application/json",

                "content-type": "application/json",

            },

            json=data,

        )

        try:

            resp.raise_for_status()

        except requests.exceptions.HTTPError as he:

            print(

                f"     ... create {AnsiColors.YELLOW}{'/'.join(project_path)}{AnsiColors.RESET} {AnsiColors.HRED}failed{AnsiColors.RESET} ({he.response.status_code}): {AnsiColors.HGRAY}{he.response.text}{AnsiColors.RESET}",

                he,

            )

            raise

        # retrieve UUID from response and return

        created_uuid = resp.json()["uuid"]

        print(f"     ... {AnsiColors.YELLOW}{'/'.join(project_path)}{AnsiColors.RESET} created: {created_uuid}")

        project_path[-1] = "#" + created_uuid

        return project_path



    def publish(self, sbom_file: Path):
@@ -261,6 +308,9 @@ class Scanner: 
                and allow_retry

            ):

                # try to create parent projects

                print(

                    f"   try to create full project path {AnsiColors.YELLOW}{'/'.join(target_project_path)}{AnsiColors.RESET}...",

                )

                target_project_path = self.create_parent_projects(target_project_path)

                # then retry

                self.do_publish(sbom_content, target_project_path, allow_retry=False)