Merge pull request #4 from OWASP/secret-auth

add custom auth type

src/.env.sample

@@ -1,2 +1,3 @@
 REDIS_HOST=localhost
 REDIS_PORT=6379
+AUTH_SECRET_KEY=E8f_3pHCgScOxTvyYFa1LKeJWLa4KtPKLY.FyLzesj66nHxQ5h1qhCFMQJJ_9eUL56EL3._XbYrYY8c5.foTV_yYHKcySPzBwv8FxIA1p03RFEinbex4EYZ9YvhacAiW

src/offat/api/app.py

@@ -1,13 +1,14 @@
-from fastapi import status, Response
+from fastapi import status, Request, Response
 from json import loads as json_loads
 from yaml import SafeLoader, load as yaml_loads
-from .config import app, task_queue, task_timeout
+from .config import app, task_queue, task_timeout, auth_secret_key
 from .jobs import scan_api
 from .models import CreateScanModel
 from ..logger import create_logger
 
 
 logger = create_logger(__name__)
+logger.info(f'Secret Key: {auth_secret_key}')
 
 
 @app.get('/', status_code=status.HTTP_200_OK)
@@ -20,9 +21,18 @@ async def root():
 
 
 @app.post('/api/v1/scan', status_code=status.HTTP_201_CREATED)
-async def add_scan_task(postData: CreateScanModel, response: Response):
-    openapi_doc = postData.openAPI
-    file_data_type = postData.type
+async def add_scan_task(scan_data: CreateScanModel, request:Request ,response: Response):
+   # for auth
+    client_ip = request.client.host
+    secret_key = request.headers.get('SECRET-KEY', None)
+    if secret_key != auth_secret_key:
+        # return 404 for better endpoint security
+        response.status_code = status.HTTP_401_UNAUTHORIZED
+        logger.warning(f'INTRUSION: {client_ip} tried to create a new scan job')
+        return {"message":"Unauthorized"}
+
+    openapi_doc = scan_data.openAPI
+    file_data_type = scan_data.type
 
     msg = {
         "msg":"Scan Task Created",
@@ -46,13 +56,26 @@ async def add_scan_task(postData: CreateScanModel, response: Response):
         job = task_queue.enqueue(scan_api, openapi_doc, job_timeout=task_timeout)
         msg['job_id'] = job.id
 
+        logger.info(f'SUCCESS: {client_ip} created new scan job - {job.id}')
+    else: 
+        logger.error(f'FAILED: {client_ip} tried creating new scan job but it failed due to unknown file data type')
+
     return msg
 
 
 @app.get('/api/v1/scan/{job_id}/result')
-async def get_scan_task_result(job_id:str, response:Response):
+async def get_scan_task_result(job_id:str, request: Request, response:Response):
+    # for auth
+    client_ip = request.client.host
+    secret_key = request.headers.get('SECRET-KEY', None)
+    if secret_key != auth_secret_key:
+        # return 404 for better endpoint security
+        response.status_code = status.HTTP_401_UNAUTHORIZED
+        logger.warning(f'INTRUSION: {client_ip} tried to access {job_id} job scan results')
+        return {"message":"Unauthorized"}
 
     scan_results = task_queue.fetch_job(job_id=job_id)
+    logger.info(f'SUCCESS: {client_ip} accessed {job_id} job scan results')
 
     msg = {
         'msg':'Task Remaining or Invalid Job Id',

src/offat/api/auth_utils.py

@@ -0,0 +1,12 @@
+import secrets
+import string
+
+
+def generate_random_secret_key_string(length=128):
+    # Define the characters allowed in the HTTP header
+    characters = string.ascii_letters + string.digits + "-_."
+
+    # Generate a random string of the specified length
+    random_string = ''.join(secrets.choice(characters) for _ in range(length))
+
+    return random_string

src/offat/api/config.py

@@ -4,12 +4,16 @@
 from dotenv import load_dotenv
 from os import environ
 
+from .auth_utils import generate_random_secret_key_string
+
+
 load_dotenv()
 
 app = FastAPI(
     title='OFFAT - API'
 )
 
+auth_secret_key = environ.get('AUTH_SECRET_KEY', generate_random_secret_key_string())
 redis_con = Redis(host=environ.get('REDIS_HOST','localhost'), port=int(environ.get('REDIS_PORT',6379)))
 task_queue = Queue(name='offat_task_queue', connection=redis_con)
 task_timeout = 60 * 60 # 3600 s = 1 hour