Support 202202 patch 1 (#1129)

* remove survey notifications

* fixing old reference of IXF_IMPORTER_DAYS_UNTIL_TICKET through EnvironmentSettings, this setting is no longer controlled through that and should come straight from settings

* Bad API keys need to return 401 just like a bad user/pass. Presently they return 200. #1117

Co-authored-by: Stefan Pratter <[email protected]>
Co-authored-by: David Poarch <[email protected]>

mainsite/settings/__init__.py

@@ -652,6 +652,7 @@ MIDDLEWARE += (
     "peeringdb_server.maintenance.Middleware",
     "peeringdb_server.middleware.CurrentRequestContext",
     "peeringdb_server.middleware.PDBCommonMiddleware",
+    "peeringdb_server.middleware.PDBPermissionMiddleware",
     "oauth2_provider.middleware.OAuth2TokenMiddleware",
 )
 

peeringdb_server/middleware.py

@@ -2,10 +2,17 @@
 Custom django middleware.
 """
 
+import base64
+
 from django.conf import settings
+from django.contrib.auth import authenticate
+from django.http import HttpResponse, JsonResponse
 from django.middleware.common import CommonMiddleware
+from django.utils.deprecation import MiddlewareMixin
 
 from peeringdb_server.context import current_request
+from peeringdb_server.models import OrganizationAPIKey, UserAPIKey
+from peeringdb_server.permissions import get_key_from_request
 
 
 class CurrentRequestContext:
@@ -24,6 +31,10 @@ class CurrentRequestContext:
             return self.get_response(request)
 
 
+class HttpResponseUnauthorized(HttpResponse):
+    status_code = 401
+
+
 class PDBCommonMiddleware(CommonMiddleware):
     def has_subdomain(self, request):
         # Check if the request has a subdomain and does not start with www
@@ -50,3 +61,90 @@ class PDBCommonMiddleware(CommonMiddleware):
         if redirect_url or path != request.get_full_path():
             redirect_url += path
             return self.response_redirect_class(redirect_url)
+
+
+class PDBPermissionMiddleware(MiddlewareMixin):
+
+    """
+    Middleware that checks if the current user has the correct permissions
+    to access the requested resource.
+    """
+
+    def get_username_and_password(self, http_auth):
+        """
+        Get the username and password from the HTTP auth header.
+        """
+        # Check if the HTTP auth header is valid.
+        if http_auth.startswith("Basic "):
+            # Get the HTTP auth header without the "Basic " prefix.
+            http_auth = http_auth[6:]
+        else:
+            # Return an empty tuple.
+            return tuple()
+        # Decode the HTTP auth header.
+        http_auth = base64.b64decode(http_auth).decode("utf-8")
+        # If username or password is empty return an empty tuple.
+        # Split the username and password from the HTTP auth header.
+        userpw = http_auth.split(":", 1)
+
+        return userpw
+
+    def response_unauthorized(self, request, status=None, message=None):
+        """
+        Return a Unauthorized response.
+        """
+        return JsonResponse({"meta": {"error": message}}, status=status)
+
+    def process_request(self, request):
+
+        http_auth = request.META.get("HTTP_AUTHORIZATION", None)
+        req_key = get_key_from_request(request)
+        api_key = None
+
+        # Check if HTTP auth is valid and if the request is made with basic auth.
+        if http_auth and http_auth.startswith("Basic "):
+            # Get the username and password from the HTTP auth header.
+            username, password = self.get_username_and_password(http_auth)
+            # Check if the username and password are valid.
+            user = authenticate(username=username, password=password)
+            # if user is not authenticated return 401 Unauthorized
+            if not user:
+
+                return self.response_unauthorized(
+                    request, message="Invalid username or password", status=401
+                )
+
+        # Check API keys
+        if req_key:
+            try:
+                api_key = OrganizationAPIKey.objects.get_from_key(req_key)
+
+            except OrganizationAPIKey.DoesNotExist:
+                pass
+
+            try:
+                api_key = UserAPIKey.objects.get_from_key(req_key)
+
+            except UserAPIKey.DoesNotExist:
+                pass
+
+            # If api key is not valid return 401 Unauthorized
+            if not api_key:
+
+                return self.response_unauthorized(
+                    request, message="Invalid API key", status=401
+                )
+
+            # If API key is provided, check if the user has an active session
+            if api_key:
+
+                if request.session.get("_auth_user_id") and request.user.id:
+                    if int(request.user.id) == int(
+                        request.session.get("_auth_user_id")
+                    ):
+
+                        return self.response_unauthorized(
+                            request,
+                            message="Cannot authenticate through Authorization header while logged in. Please log out and try again.",
+                            status=400,
+                        )

tests/test_keys.py

@@ -1,6 +1,8 @@
+from lib2to3.pgen2 import token
+
 import pytest
 from django.conf import settings
-from django.test import RequestFactory
+from django.test import Client, RequestFactory
 from django.urls import reverse
 from django_grainy.models import UserPermission
 from grainy.const import PERM_CRUD, PERM_READ
@@ -339,3 +341,33 @@ def test_get_network_w_user_key(network, user, org):
     assert net_from_api["name"] == network.name
     assert net_from_api["asn"] == network.asn
     assert net_from_api["org_id"] == network.org.id
+
+
+@pytest.mark.django_db
+def test_bogus_api_key(network):
+    url = reverse("net-detail", args=(network.id,))
+    key = "abcd"
+    client = APIClient()
+    client.credentials(HTTP_AUTHORIZATION="Api-Key " + key)
+    client.force_authenticate(token=key)
+    response = client.get(url)
+    assert response.status_code == 401
+
+
+@pytest.mark.django_db
+def test_key_active_user_session(network, user, org):
+    namespace = f"peeringdb.organization.{org.id}.network"
+    userperm = UserPermission.objects.create(
+        namespace=namespace, permission=PERM_CRUD, user=user
+    )
+    api_key, key = UserAPIKey.objects.create_key(name="test key", user=user)
+
+    url = reverse("net-detail", args=(network.id,))
+    client = APIClient()
+    client.credentials(HTTP_AUTHORIZATION="Api-Key " + key)
+    client.login(username="user", password="user")
+    # Save session for user
+    client.session["_auth_user_id"] = user.id
+    client.session.save()
+    response = client.get(url)
+    assert response.status_code == 400

tests/test_views.py

@@ -1,3 +1,4 @@
+import base64
 import re
 
 import pytest
@@ -196,3 +197,12 @@ def test_close_account():
     assert user.email == ""
     assert user.first_name == ""
     assert user.last_name == ""
+
+
+@pytest.mark.django_db
+def test_bogus_basic_auth():
+    auth_string = "Basic YmFkOmJhZA=="
+    auth_headers = {"HTTP_AUTHORIZATION": auth_string}
+    client = Client()
+    response = client.get("/", **auth_headers)
+    assert response.status_code == 401