Implement Blocklist checking system

backend/src/xfd_django/xfd_api/api_methods/blocklist.py

@@ -0,0 +1,21 @@
+"""Blocklist API."""
+
+# Third-Party Libraries
+from fastapi import HTTPException
+from xfd_mini_dl.models import Blocklist
+
+
+async def handle_check_ip(ip_address: str):
+    """
+    Determine if an IP exists in our blocklist table.
+
+    Returns:
+        { status: "BLOCKED" or "UNBLOCKED" }
+    """
+    try:
+        Blocklist.objects.get(ip=ip_address)
+        return {"status": "BLOCKED"}
+    except Blocklist.DoesNotExist:
+        return {"status": "UNBLOCKED"}
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))

backend/src/xfd_django/xfd_api/schema_models/blocklist.py

@@ -0,0 +1,9 @@
+"""Blocklist Schemas."""
+# Third-Party Libraries
+from pydantic import BaseModel
+
+
+class BlocklistCheckResponse(BaseModel):
+    """BlocklistCheckResponse schema."""
+
+    status: str

backend/src/xfd_django/xfd_api/schema_models/scan.py

@@ -317,6 +317,15 @@ class GenericMessageResponseModel(BaseModel):
         memory="16384",
         description="Loops through all domains and determines if their associated IP can be found in a report Cidr block.",
     ),
+    "updateBlocklist": ScanSchema(
+        type="fargate",
+        isPassive=True,
+        global_scan=True,
+        numChunks=0,
+        cpu="1024",
+        memory="8192",
+        description="Updates blocked ip records against blocklist.de global IP blocklist",
+    ),
     "was_sync": ScanSchema(
         type="fargate",
         isPassive=True,

backend/src/xfd_django/xfd_api/tests/test_blocklist.py

@@ -0,0 +1,59 @@
+"""Test Blocklist Check."""
+
+# Standard Library
+# Standard Python Libraries
+from datetime import datetime, timezone
+import secrets
+
+# Third-Party Libraries
+from fastapi.testclient import TestClient
+import pytest
+from xfd_api.auth import create_jwt_token
+from xfd_api.models import User, UserType
+from xfd_django.asgi import app
+from xfd_mini_dl.models import Blocklist
+
+client = TestClient(app)
+
+
+@pytest.mark.django_db(transaction=True, databases=["default", "mini_data_lake"])
+def test_blocklist_check_blocked():
+    """Test blocklist check."""
+    user = User.objects.create(
+        firstName="first",
+        lastName="last",
+        email="{}@crossfeed.cisa.gov".format(secrets.token_hex(4)),
+        userType=UserType.STANDARD,
+    )
+    random_ip_address = "111.111.111.111"
+    Blocklist.objects.create(
+        ip=random_ip_address, created_at=datetime.now(timezone.utc)
+    )
+
+    response = client.get(
+        "/blocklist/check/",
+        params={"ip_address": random_ip_address},
+        headers={"Authorization": "Bearer {}".format(create_jwt_token(user))},
+    )
+
+    assert response.status_code == 200
+    assert response.json() == {"status": "BLOCKED"}
+
+
+@pytest.mark.django_db(transaction=True, databases=["default", "mini_data_lake"])
+def test_blocklist_check_unblocked():
+    """Test blocklist check."""
+    user = User.objects.create(
+        firstName="first",
+        lastName="last",
+        email="{}@crossfeed.cisa.gov".format(secrets.token_hex(4)),
+        userType=UserType.STANDARD,
+    )
+    random_ip_address = "222.222.222.222"
+    response = client.get(
+        "/blocklist/check/",
+        params={"ip_address": random_ip_address},
+        headers={"Authorization": "Bearer {}".format(create_jwt_token(user))},
+    )
+
+    assert response.json() == {"status": "UNBLOCKED"}

backend/src/xfd_django/xfd_api/views.py

@@ -14,6 +14,7 @@
 from .api_methods import api_key as api_key_methods
 from .api_methods import notification as notification_methods
 from .api_methods import organization, proxy, scan, scan_tasks, user
+from .api_methods.blocklist import handle_check_ip
 from .api_methods.cpe import get_cpes_by_id
 from .api_methods.cve import get_cves_by_id, get_cves_by_name
 from .api_methods.domain import export_domains, get_domain_by_id, search_domains
@@ -61,6 +62,7 @@
 from .schema_models import scan_tasks as scanTaskSchema
 from .schema_models import stat_schema
 from .schema_models.api_key import ApiKey as ApiKeySchema
+from .schema_models.blocklist import BlocklistCheckResponse
 from .schema_models.cpe import Cpe as CpeSchema
 from .schema_models.cve import Cve as CveSchema
 from .schema_models.domain import DomainSearch, DomainSearchResponse, GetDomainResponse
@@ -1360,3 +1362,21 @@ async def call_update_vulnerability(
         object: a single vulnerability object that has been modified.
     """
     return update_vulnerability(vulnerability_id, data, current_user)
+
+
+# ========================================
+#   Blocklist Endpoints
+# ========================================
+
+
+@api_router.get(
+    "/blocklist/check",
+    dependencies=[Depends(get_current_active_user)],
+    response_model=BlocklistCheckResponse,
+    tags=["Blocklist"],
+)
+async def get_blocklist(
+    request: Request, ip_address: str = Query(..., description="IP address to check")
+):
+    """Determine if IP is on the blocklist."""
+    return await handle_check_ip(ip_address)

backend/src/xfd_django/xfd_mini_dl/models.py

@@ -5640,6 +5640,27 @@ class Meta:
         unique_together = (("cpe_product_name", "version_number"),)
 
 
+class Blocklist(models.Model):
+    """Define Blocklist Model."""
+
+    id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
+    ip = InetAddressField(
+        null=False, blank=False, unique=True
+    )  # <-- Removed trailing comma
+    created_at = models.DateTimeField(auto_now=False)  # <-- Removed trailing comma
+
+    class Meta:
+        """Set Blocklist model metadata."""
+
+        app_label = app_label_name
+        managed = manage_db
+        db_table = "blocklist"
+        indexes = [
+            models.Index(fields=["ip"]),  # Reinforces index on 'ip' field
+            models.Index(fields=["created_at"]),  # Speeds up sorting by 'updated_at'
+        ]
+
+
 # # THese are all views, so they shouldn't be generated via the ORM
 
 # # This should be a view not a table
@@ -6398,5 +6419,6 @@ class Meta:
 #     class Meta:
 #         """Set VwIscoreOrgsIpCounts model metadata."""
 
+
 #         managed = False
 #         db_table = "vw_iscore_orgs_ip_counts"""" Django ORM models """