SocketDev · dacoburn · Jun 3, 2024 · May 31, 2024
@@ -2,5 +2,5 @@ FROM python:3-alpine
 LABEL org.opencontainers.image.authors="socket.dev"
 
 RUN apk update \
-    && apk add --no-cache git
-RUN pip install socketsecurity
+    && apk add --no-cache git nodejs npm yarn
+RUN pip install socketsecurity --upgrade
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "socketsecurity"
-version = "0.0.67"
+version = "0.0.72"
 requires-python = ">= 3.9"
 dependencies = [
     'requests',

@@ -25,7 +25,7 @@
 
 
 __author__ = 'socket.dev'
-__version__ = '0.0.67'
+__version__ = '0.0.72'
 __all__ = [
     "Core",
     "log",
@@ -246,6 +246,55 @@ def get_security_policy() -> dict:
                 }
         return org_rules
 
+    # @staticmethod
+    # def get_supported_file_types() -> dict:
+    #     path = "report/supported"
+
+    @staticmethod
+    def get_manifest_files(package: Package, packages: dict) -> str:
+        if package.direct:
+            manifests = []
+            for manifest_item in package.manifestFiles:
+                manifest = manifest_item["file"]
+                manifests.append(manifest)
+            manifest_files = ";".join(manifests)
+        else:
+            manifests = []
+            for top_id in package.topLevelAncestors:
+                top_package: Package
+                top_package = packages[top_id]
+                for manifest_item in top_package.manifestFiles:
+                    manifest = manifest_item["file"]
+                    new_string = f"{package.name}@{package.version}({manifest})"
+                    manifests.append(new_string)
+            manifest_files = ";".join(manifests)
+        return manifest_files
+
+    @staticmethod
+    def create_sbom_output(diff: Diff) -> list:
+        sbom = []
+        for package_id in diff.packages:
+            package: Package
+            package = diff.packages[package_id]
+            manifest_files = Core.get_manifest_files(package, diff.packages)
+            item = {
+                "id": package.id,
+                "license": package.license,
+                "license_text": package.license_text,
+                "manifestFiles": manifest_files,
+                "score": package.score,
+                "size": package.size,
+                "ecosystem": package.type,
+                "alerts": package.alerts,
+                "direct": package.direct,
+                "name": package.name,
+                "version": package.version,
+                "author": package.author,
+                "url": package.url
+            }
+            sbom.append(item)
+        return sbom
+
     @staticmethod
     def find_files(path: str) -> list:
         """
@@ -314,8 +363,8 @@ def find_files(path: str) -> list:
                 "requirements.frozen": {
                     "pattern": "requirements.frozen"
                 },
-                "setup.py.old": {
-                    "pattern": "setup.py.old"
+                "setup.py": {
+                    "pattern": "setup.py"
                 }
             },
             "golang": {
@@ -325,6 +374,11 @@ def find_files(path: str) -> list:
                 "go.sum": {
                     "pattern": "go.sum"
                 }
+            },
+            "java": {
+                "pom.xml": {
+                    "pattern": "pom.xml"
+                }
             }
         }
         all_files = []
@@ -469,12 +523,12 @@ def compare_sboms(new_scan: list, head_scan: list) -> Diff:
 
         for package_id in new_packages:
             purl, package = Core.create_purl(package_id, new_packages)
-            if package_id not in head_packages:
+            if package_id not in head_packages and package.direct:
                 diff.new_packages.append(purl)
             new_scan_alerts = Core.create_issue_alerts(package, new_scan_alerts, new_packages)
         for package_id in head_packages:
             purl, package = Core.create_purl(package_id, head_packages)
-            if package_id not in new_packages:
+            if package_id not in new_packages and package.direct:
                 diff.removed_packages.append(purl)
             head_scan_alerts = Core.create_issue_alerts(package, head_scan_alerts, head_packages)
         diff.new_alerts = Core.compare_issue_alerts(new_scan_alerts, head_scan_alerts, diff.new_alerts)

@@ -163,19 +163,27 @@ def check_event_type() -> str:
         return event_type
 
     @staticmethod
-    def add_socket_comments(security_comment: str, overview_comment: str, comments: dict) -> None:
+    def add_socket_comments(
+            security_comment: str,
+            overview_comment: str,
+            comments: dict,
+            new_security_comment: bool = True,
+            new_overview_comment: bool = True
+    ) -> None:
         existing_overview_comment = comments.get("overview")
         existing_security_comment = comments.get("security")
-        if existing_overview_comment is not None:
-            existing_overview_comment: GithubComment
-            Github.update_comment(overview_comment, str(existing_overview_comment.id))
-        else:
-            Github.post_comment(overview_comment)
-        if existing_security_comment is not None:
-            existing_security_comment: GithubComment
-            Github.update_comment(security_comment, str(existing_security_comment.id))
-        else:
-            Github.post_comment(security_comment)
+        if new_overview_comment:
+            if existing_overview_comment is not None:
+                existing_overview_comment: GithubComment
+                Github.update_comment(overview_comment, str(existing_overview_comment.id))
+            else:
+                Github.post_comment(overview_comment)
+        if new_security_comment:
+            if existing_security_comment is not None:
+                existing_security_comment: GithubComment
+                Github.update_comment(security_comment, str(existing_security_comment.id))
+            else:
+                Github.post_comment(security_comment)
 
     @staticmethod
     def post_comment(body: str) -> None:

@@ -141,7 +141,7 @@ def __init__(self):
         self.api_token = gitlab_token
         self.project_id = ci_merge_request_project_id
         if self.api_token is None:
-            print("Unable to get gitlab API Token from GH_API_TOKEN")
+            print("Unable to get gitlab API Token from GITLAB_TOKEN")
             sys.exit(2)
 
     @staticmethod
@@ -159,19 +159,27 @@ def check_event_type() -> str:
         return event_type
 
     @staticmethod
-    def add_socket_comments(security_comment: str, overview_comment: str, comments: dict) -> None:
+    def add_socket_comments(
+            security_comment: str,
+            overview_comment: str,
+            comments: dict,
+            new_security_comment: bool = True,
+            new_overview_comment: bool = True
+    ) -> None:
         existing_overview_comment = comments.get("overview")
         existing_security_comment = comments.get("security")
-        if existing_overview_comment is not None:
-            existing_overview_comment: GitlabComment
-            Gitlab.update_comment(overview_comment, str(existing_overview_comment.id))
-        else:
-            Gitlab.post_comment(overview_comment)
-        if existing_security_comment is not None:
-            existing_security_comment: GitlabComment
-            Gitlab.update_comment(security_comment, str(existing_security_comment.id))
-        else:
-            Gitlab.post_comment(security_comment)
+        if new_overview_comment:
+            if existing_overview_comment is not None:
+                existing_overview_comment: GitlabComment
+                Gitlab.update_comment(overview_comment, str(existing_overview_comment.id))
+            else:
+                Gitlab.post_comment(overview_comment)
+        if new_security_comment:
+            if existing_security_comment is not None:
+                existing_security_comment: GitlabComment
+                Gitlab.update_comment(security_comment, str(existing_security_comment.id))
+            else:
+                Gitlab.post_comment(security_comment)
 
     @staticmethod
     def post_comment(body: str) -> None:

@@ -69,6 +69,20 @@
     required=False
 )
 
+parser.add_argument(
+    '--sbom-file',
+    default=None,
+    help='If soecified save the SBOM details to the specified file',
+    required=False
+)
+
+parser.add_argument(
+    '--commit-sha',
+    default="",
+    help='Optional git commit sha',
+    required=False
+)
+
 parser.add_argument(
     '--generate-license',
     default=False,
@@ -115,6 +129,8 @@ def cli():
     pr_number = arguments.pr_number
     target_path = arguments.target_path
     scm_type = arguments.scm
+    commit_sha = arguments.commit_sha
+    sbom_file = arguments.sbom_file
     license_mode = arguments.generate_license
     license_file = f"{repo}"
     if branch is not None:
@@ -137,6 +153,7 @@ def cli():
         scm = Gitlab()
     if scm is not None:
         default_branch = scm.is_default_branch
+
     base_api_url = os.getenv("BASE_API_URL") or None
     core = Core(token=api_token, request_timeout=6000, base_api_url=base_api_url)
     set_as_pending_head = False
@@ -146,7 +163,7 @@ def cli():
         repo=repo,
         branch=branch,
         commit_message=commit_message,
-        commit_hash="",
+        commit_hash=commit_sha,
         pull_request=pr_number,
         committers=committer,
         make_default_branch=default_branch,
@@ -166,7 +183,19 @@ def cli():
             diff.new_alerts = scm.remove_alerts(comments, diff.new_alerts)
             overview_comment = Messages.dependency_overview_template(diff)
             security_comment = Messages.security_comment_template(diff)
-            scm.add_socket_comments(security_comment, overview_comment, comments)
+            new_security_comment = True
+            new_overview_comment = True
+            if len(diff.new_alerts) == 0:
+                new_security_comment = False
+            if len(diff.new_packages) == 0 and diff.removed_packages == 0:
+                new_overview_comment = False
+            scm.add_socket_comments(
+                security_comment,
+                overview_comment,
+                comments,
+                new_security_comment,
+                new_overview_comment
+            )
         output_console_comments(diff)
     else:
         log.info("API Mode")
@@ -190,6 +219,8 @@ def cli():
             }
             all_packages[package_id] = output
         core.save_file(license_file, json.dumps(all_packages))
+    if diff is not None and sbom_file is not None:
+        core.save_file(sbom_file, json.dumps(core.create_sbom_output(diff)))
 
 
 if __name__ == '__main__':

@@ -0,0 +1,35 @@
+import requests
+import base64
+from urllib.parse import urlencode
+import time
+api_token = "sktsec_b1q-gQvQXJMR0ZtfzPgmQx_PZNRhxEsxJUoOMK7rYhzM_api"
+token = f"{api_token}:"
+encoded_token = base64.b64encode(token.encode()).decode('ascii')
+
+url = 'https://api.socket.dev/v0/orgs/socketdev-demo/full-scans'
+params = {
+    'repo': 'new-local-test',
+    'branch': 'test'
+}
+full_url = f"{url}?{urlencode(params)}"
+headers = {
+    'Accept': 'application/json',
+    'Authorization': f'Basic {encoded_token}'
+}
+key = 'requirements.txt'
+files = []
+file = (
+    key,
+    (
+        key,
+        open(key, 'rb')
+    )
+)
+files.append(file)
+
+resp = requests.post(full_url, headers=headers, files=files)
+
+print(resp.status_code)
+print(resp.text)
+
+time.sleep(10)