aboutcode-org · TG1999 · Aug 17, 2022 · Aug 17, 2022 · Aug 17, 2022 · Aug 17, 2022
diff --git a/CHANGELOG.rst b/CHANGELOG.rst
@@ -59,6 +59,13 @@ Version v30.0.0
 
 - Add bulk search support for CPEs.
 
+- Add authentication for REST API endpoint.
+  The autentication is disabled by default and can be enabled using the
+  SCANCODEIO_REQUIRE_AUTHENTICATION settings.
+  When enabled, users have to authenticate using 
+  their API Key in the REST API.
+  Users can be created using the Django "createsuperuser" management command.
+
 Other:
 
 - we dropped calver to use a plain semver.

diff --git a/vulnerabilities/models.py b/vulnerabilities/models.py
@@ -12,13 +12,16 @@
 import logging
 import uuid
 
+from django.conf import settings
 from django.core.exceptions import ValidationError
 from django.core.validators import MaxValueValidator
 from django.core.validators import MinValueValidator
 from django.db import models
+from django.dispatch import receiver
 from django.utils.http import int_to_base36
 from packageurl import PackageURL
 from packageurl.contrib.django.models import PackageURLMixin
+from rest_framework.authtoken.models import Token
 
 from vulnerabilities.importer import AdvisoryData
 from vulnerabilities.importer import AffectedPackage
@@ -418,3 +421,12 @@ def to_advisory_data(self) -> AdvisoryData:
             references=[Reference.from_dict(ref) for ref in self.references],
             date_published=self.date_published,
         )
+
+
+@receiver(models.signals.post_save, sender=settings.AUTH_USER_MODEL)
+def create_auth_token(sender, instance=None, created=False, **kwargs):
+    """
+    Creates an API key token on user creation, using the signal system.
+    """
+    if created:
+        Token.objects.create(user_id=instance.pk)
diff --git a/vulnerabilities/templates/base.html b/vulnerabilities/templates/base.html
@@ -1,4 +1,5 @@
-<!DOCTYPE html class="has-navbar-fixed-top">
+<!DOCTYPE html>
+<html lang="en">
 {% load static %}
 <head>
     <title>VulnerableCode</title>

diff --git a/vulnerabilities/tests/test_auth.py b/vulnerabilities/tests/test_auth.py
@@ -0,0 +1,34 @@
+#
+# Copyright (c) nexB Inc. and others. All rights reserved.
+# VulnerableCode is a trademark of nexB Inc.
+# SPDX-License-Identifier: Apache-2.0
+# See http://www.apache.org/licenses/LICENSE-2.0 for the license text.
+# See https://github.com/nexB/vulnerablecode for support or download.
+# See https://aboutcode.org for more information about nexB OSS projects.
+# This is copied from https://github.com/nexB/scancode.io/commit/eab8eeb13989c26a1600cc64e8b054f171341063
+#
+
+
+from django.conf import settings
+from django.contrib.auth import get_user_model
+from django.contrib.auth.models import AnonymousUser
+from django.test import TestCase
+
+TEST_PASSWORD = "secret"
+
+User = get_user_model()
+
+api_package_url = "/api/packages/"
+login_redirect_url = settings.LOGIN_REDIRECT_URL
+
+
+class VulnerableCodeAuthTest(TestCase):
+    def setUp(self):
+        self.anonymous_user = AnonymousUser()
+        self.basic_user = User.objects.create_user(username="basic_user", password=TEST_PASSWORD)
+
+    def test_vulnerablecode_auth_api_required_authentication(self):
+        response = self.client.get(api_package_url)
+        expected = {"detail": "Authentication credentials were not provided."}
+        self.assertEqual(expected, response.json())
+        self.assertEqual(401, response.status_code)
diff --git a/vulnerabilities/tests/test_fix_api.py b/vulnerabilities/tests/test_fix_api.py
@@ -7,10 +7,15 @@
 # See https://aboutcode.org for more information about nexB OSS projects.
 #
 
+import json
+
+from django.contrib.auth import get_user_model
 from django.test import TestCase
+from django.test import TransactionTestCase
 from django.utils.http import int_to_base36
 from packageurl import PackageURL
 from rest_framework import status
+from rest_framework.test import APIClient
 
 from vulnerabilities.models import Alias
 from vulnerabilities.models import Package
@@ -19,9 +24,15 @@
 from vulnerabilities.models import VulnerabilityReference
 from vulnerabilities.models import VulnerabilityRelatedReference
 
+User = get_user_model()
+
 
-class APITestCaseVulnerability(TestCase):
+class APITestCaseVulnerability(TransactionTestCase):
     def setUp(self):
+        self.user = User.objects.create_user("username", "e@mail.com", "secret")
+        self.auth = f"Token {self.user.auth_token.key}"
+        self.csrf_client = APIClient(enforce_csrf_checks=True)
+        self.csrf_client.credentials(HTTP_AUTHORIZATION=self.auth)
         for i in range(0, 200):
             Vulnerability.objects.create(
                 summary=str(i),
@@ -35,15 +46,15 @@ def setUp(self):
             )
 
     def test_api_status(self):
-        response = self.client.get("/api/vulnerabilities/", format="json")
+        response = self.csrf_client.get("/api/vulnerabilities/")
         self.assertEqual(status.HTTP_200_OK, response.status_code)
 
     def test_api_response(self):
-        response = self.client.get("/api/vulnerabilities/", format="json").data
+        response = self.csrf_client.get("/api/vulnerabilities/").data
         self.assertEqual(response["count"], 201)
 
     def test_api_with_single_vulnerability(self):
-        response = self.client.get(
+        response = self.csrf_client.get(
             f"/api/vulnerabilities/{self.vulnerability.id}", format="json"
         ).data
         assert response == {
@@ -66,7 +77,7 @@ def test_api_with_single_vulnerability(self):
         }
 
     def test_api_with_single_vulnerability_with_filters(self):
-        response = self.client.get(
+        response = self.csrf_client.get(
             f"/api/vulnerabilities/{self.vulnerability.id}?type=pypi", format="json"
         ).data
         assert response == {
@@ -87,6 +98,10 @@ def test_api_with_single_vulnerability_with_filters(self):
 
 class APITestCasePackage(TestCase):
     def setUp(self):
+        self.user = User.objects.create_user("username", "e@mail.com", "secret")
+        self.auth = f"Token {self.user.auth_token.key}"
+        self.csrf_client = APIClient(enforce_csrf_checks=True)
+        self.csrf_client.credentials(HTTP_AUTHORIZATION=self.auth)
         vuln = Vulnerability.objects.create(
             summary="test-vuln",
         )
@@ -123,15 +138,15 @@ def setUp(self):
         )
 
     def test_api_status(self):
-        response = self.client.get("/api/packages/", format="json")
+        response = self.csrf_client.get("/api/packages/", format="json")
         self.assertEqual(status.HTTP_200_OK, response.status_code)
 
     def test_api_response(self):
-        response = self.client.get("/api/packages/", format="json").data
+        response = self.csrf_client.get("/api/packages/", format="json").data
         self.assertEqual(response["count"], 11)
 
     def test_api_with_single_vulnerability_and_fixed_package(self):
-        response = self.client.get(f"/api/packages/{self.package.id}", format="json").data
+        response = self.csrf_client.get(f"/api/packages/{self.package.id}", format="json").data
         assert response == {
             "url": f"http://testserver/api/packages/{self.package.id}",
             "purl": "pkg:generic/nginx/test@11",
@@ -160,7 +175,7 @@ def test_api_with_single_vulnerability_and_fixed_package(self):
         }
 
     def test_api_with_single_vulnerability_and_vulnerable_package(self):
-        response = self.client.get(f"/api/packages/{self.vuln_package.id}", format="json").data
+        response = self.csrf_client.get(f"/api/packages/{self.vuln_package.id}", format="json").data
         assert response == {
             "url": f"http://testserver/api/packages/{self.vuln_package.id}",
             "purl": "pkg:generic/nginx/test@9",
@@ -204,6 +219,10 @@ def test_api_with_single_vulnerability_and_vulnerable_package(self):
 
 class CPEApi(TestCase):
     def setUp(self):
+        self.user = User.objects.create_user("username", "e@mail.com", "secret")
+        self.auth = f"Token {self.user.auth_token.key}"
+        self.csrf_client = APIClient(enforce_csrf_checks=True)
+        self.csrf_client.credentials(HTTP_AUTHORIZATION=self.auth)
         self.vulnerability = Vulnerability.objects.create(summary="test")
         for i in range(0, 10):
             ref, _ = VulnerabilityReference.objects.get_or_create(
@@ -214,31 +233,39 @@ def setUp(self):
             )
 
     def test_api_status(self):
-        response = self.client.get("/api/cpes/", format="json")
+        response = self.csrf_client.get("/api/cpes/", format="json")
         self.assertEqual(status.HTTP_200_OK, response.status_code)
 
     def test_api_response(self):
-        response = self.client.get("/api/cpes/?cpe=cpe:/a:nginx:9", format="json").data
+        response = self.csrf_client.get("/api/cpes/?cpe=cpe:/a:nginx:9", format="json").data
         self.assertEqual(response["count"], 1)
 
 
 class AliasApi(TestCase):
     def setUp(self):
+        self.user = User.objects.create_user("username", "e@mail.com", "secret")
+        self.auth = f"Token {self.user.auth_token.key}"
+        self.csrf_client = APIClient(enforce_csrf_checks=True)
+        self.csrf_client.credentials(HTTP_AUTHORIZATION=self.auth)
         self.vulnerability = Vulnerability.objects.create(summary="test")
         for i in range(0, 10):
             Alias.objects.create(alias=f"CVE-{i}", vulnerability=self.vulnerability)
 
     def test_api_status(self):
-        response = self.client.get("/api/alias/", format="json")
+        response = self.csrf_client.get("/api/alias/", format="json")
         self.assertEqual(status.HTTP_200_OK, response.status_code)
 
     def test_api_response(self):
-        response = self.client.get("/api/alias?alias=CVE-9", format="json").data
+        response = self.csrf_client.get("/api/alias?alias=CVE-9", format="json").data
         self.assertEqual(response["count"], 1)
 
 
-class BulkSearchAPI(TestCase):
+class BulkSearchAPIPackage(TestCase):
     def setUp(self):
+        self.user = User.objects.create_user("username", "e@mail.com", "secret")
+        self.auth = f"Token {self.user.auth_token.key}"
+        self.csrf_client = APIClient(enforce_csrf_checks=True)
+        self.csrf_client.credentials(HTTP_AUTHORIZATION=self.auth)
         packages = [
             "pkg:nginx/nginx@0.6.18",
             "pkg:nginx/nginx@1.20.0",
@@ -264,16 +291,20 @@ def test_api_response(self):
         request_body = {
             "purls": self.packages,
         }
-        response = self.client.post(
+        response = self.csrf_client.post(
             "/api/packages/bulk_search",
-            data=request_body,
+            data=json.dumps(request_body),
             content_type="application/json",
         ).json()
         assert len(response) == 13
 
 
-class BulkSearchAPI(TestCase):
+class BulkSearchAPICPE(TestCase):
     def setUp(self):
+        self.user = User.objects.create_user("username", "e@mail.com", "secret")
+        self.auth = f"Token {self.user.auth_token.key}"
+        self.csrf_client = APIClient(enforce_csrf_checks=True)
+        self.csrf_client.credentials(HTTP_AUTHORIZATION=self.auth)
         self.exclusive_cpes = [
             "cpe:/a:nginx:1.0.7",
             "cpe:/a:nginx:1.0.15",
@@ -305,9 +336,9 @@ def test_api_response_with_with_exclusive_cpes_associated_with_two_vulnerabiliti
         request_body = {
             "cpes": self.exclusive_cpes,
         }
-        response = self.client.post(
+        response = self.csrf_client.post(
             "/api/cpes/bulk_search",
-            data=request_body,
+            data=json.dumps(request_body),
             content_type="application/json",
         ).json()
         assert len(response) == 1
@@ -320,9 +351,9 @@ def test_api_response_with_no_cpe_associated(self):
         request_body = {
             "cpes": ["cpe:/a:nginx:1.10.7"],
         }
-        response = self.client.post(
+        response = self.csrf_client.post(
             "/api/cpes/bulk_search",
-            data=request_body,
+            data=json.dumps(request_body),
             content_type="application/json",
         ).json()
         assert len(response) == 0
@@ -331,9 +362,9 @@ def test_api_response_with_with_non_exclusive_cpes_associated_with_two_vulnerabi
         request_body = {
             "cpes": self.non_exclusive_cpes,
         }
-        response = self.client.post(
+        response = self.csrf_client.post(
             "/api/cpes/bulk_search",
-            data=request_body,
+            data=json.dumps(request_body),
             content_type="application/json",
         ).json()
         assert len(response) == 2
@@ -342,20 +373,18 @@ def test_with_empty_list(self):
         request_body = {
             "cpes": [],
         }
-        response = self.client.post(
+        response = self.csrf_client.post(
             "/api/cpes/bulk_search",
-            data=request_body,
+            data=json.dumps(request_body),
             content_type="application/json",
         ).json()
         assert response == {"Error": "A non-empty 'cpe' list of package URLs is required."}
 
     def test_with_invalid_cpes(self):
-        request_body = {
-            "cpes": ["CVE-2022-2022"],
-        }
-        response = self.client.post(
+        request_body = {"cpes": ["CVE-2022-2022"]}
+        response = self.csrf_client.post(
             "/api/cpes/bulk_search",
-            data=request_body,
+            data=json.dumps(request_body),
             content_type="application/json",
         ).json()
         assert response == {"Error": "Invalid CPE: CVE-2022-2022"}
diff --git a/vulnerablecode/settings.py b/vulnerablecode/settings.py
@@ -7,6 +7,7 @@
 # See https://aboutcode.org for more information about nexB OSS projects.
 #
 
+import sys
 from pathlib import Path
 
 import environ
@@ -129,6 +130,21 @@
 
 USE_I18N = True
 
+IS_TESTS = False
+
+if len(sys.argv) > 0:
+    IS_TESTS = "pytest" in sys.argv[0]
+
+VULNERABLECODEIO_REQUIRE_AUTHENTICATION = env.bool(
+    "VULNERABLECODEIO_REQUIRE_AUTHENTICATION", default=False
+)
+
+LOGIN_REDIRECT_URL = "/"
+LOGOUT_REDIRECT_URL = "/"
+
+if IS_TESTS:
+    VULNERABLECODEIO_REQUIRE_AUTHENTICATION = True
+
 USE_L10N = True
 
 USE_TZ = True
@@ -148,8 +164,8 @@
 # Django restframework
 
 REST_FRAMEWORK = {
-    "DEFAULT_AUTHENTICATION_CLASSES": ("rest_framework.authentication.SessionAuthentication",),
-    "DEFAULT_PERMISSION_CLASSES": ("rest_framework.permissions.AllowAny",),
+    "DEFAULT_AUTHENTICATION_CLASSES": ("rest_framework.authentication.TokenAuthentication",),
+    "DEFAULT_PERMISSION_CLASSES": ("rest_framework.permissions.IsAuthenticated",),
     "DEFAULT_RENDERER_CLASSES": (
         "rest_framework.renderers.JSONRenderer",
         "rest_framework.renderers.BrowsableAPIRenderer",
@@ -163,3 +179,6 @@
     # Limit the load on the Database returning a small number of records by default. https://github.com/nexB/vulnerablecode/issues/819
     "PAGE_SIZE": 10,
 }
+
+if not VULNERABLECODEIO_REQUIRE_AUTHENTICATION:
+    REST_FRAMEWORK["DEFAULT_PERMISSION_CLASSES"] = ("rest_framework.permissions.AllowAny",)