Dev RELEASE: v0.14.1

.gitignore

@@ -203,3 +203,4 @@ swagger.json
 
 ## unknown data
 .DS_Store
+oas.yml

.pre-commit-config.yaml

@@ -0,0 +1,7 @@
+repos:
+  - repo: https://github.com/ambv/black
+    rev: 24.1.1
+    hooks:
+      - id: black
+        language_version: python3.11
+        args: ["--line-length", "100", "--skip-string-normalization"]

src/offat/__main__.py

@@ -76,7 +76,6 @@ def start():
         rate_limit=rate_limit,
         test_data_config=test_data_config,
         proxy=args.proxy,
-        ssl=args.no_ssl,
     )
 
 

src/offat/api/app.py

@@ -6,11 +6,11 @@
 from os import uname, environ
 
 
-logger.info(f'Secret Key: {auth_secret_key}')
+logger.info('Secret Key: %s', auth_secret_key)
 
 
-if uname().sysname == 'Darwin' and environ.get('OBJC_DISABLE_INITIALIZE_FORK_SAFETY') != 'YES':
-    logger.warning('Mac Users might need to configure OBJC_DISABLE_INITIALIZE_FORK_SAFETY=YES in env\nVisit StackOverFlow link for more info: https://stackoverflow.com/questions/50168647/multiprocessing-causes-python-to-crash-and-gives-an-error-may-have-been-in-progr')
+# if uname().sysname == 'Darwin' and environ.get('OBJC_DISABLE_INITIALIZE_FORK_SAFETY') != 'YES':
+# logger.warning('Mac Users might need to configure OBJC_DISABLE_INITIALIZE_FORK_SAFETY=YES in env\nVisit StackOverFlow link for more info: https://stackoverflow.com/questions/50168647/multiprocessing-causes-python-to-crash-and-gives-an-error-may-have-been-in-progr')
 
 
 @app.get('/', status_code=status.HTTP_200_OK)
@@ -30,8 +30,7 @@
     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')
+        logger.warning('INTRUSION: %s tried to create a new scan job', client_ip)
         return {"message": "Unauthorized"}
 
     msg = {
@@ -42,7 +41,7 @@
     job = task_queue.enqueue(scan_api, scan_data,  job_timeout=task_timeout)
     msg['job_id'] = job.id
 
-    logger.info(f'SUCCESS: {client_ip} created new scan job - {job.id}')
+    logger.info('SUCCESS: %s created new scan job - %s', client_ip, job.id)
 
     return msg
 
@@ -55,13 +54,12 @@
     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')
+        logger.warning('INTRUSION: %s tried to access %s job scan results', client_ip, job_id)
         return {"message": "Unauthorized"}
 
     scan_results_job = task_queue.fetch_job(job_id=job_id)
 
-    logger.info(f'SUCCESS: {client_ip} accessed {job_id} job scan results')
+    logger.info('SUCCESS: %s accessed %s job scan results', client_ip, job_id)
 
     msg = 'Task Remaining or Invalid Job Id'
     results = None

src/offat/api/jobs.py

@@ -1,4 +1,4 @@
-from traceback import print_exception
+from sys import exc_info
 from offat.api.models import CreateScanModel
 from offat.tester.tester_utils import generate_and_run_tests
 from offat.openapi import OpenAPIParser
@@ -7,7 +7,6 @@
 
 def scan_api(body_data: CreateScanModel):
     try:
-        logger.info('test')
         api_parser = OpenAPIParser(fpath_or_url=None, spec=body_data.openAPI)
 
         results = generate_and_run_tests(
@@ -20,6 +19,6 @@ def scan_api(body_data: CreateScanModel):
         )
         return results
     except Exception as e:
-        logger.error(f'Error occurred while creating a job: {e}')
-        print_exception(e)
+        logger.error('Error occurred while creating a job: %s', repr(e))
+        logger.debug("Debug Data:", exc_info=exc_info())
         return [{'error': str(e)}]

src/offat/config_data_handler.py

@@ -8,8 +8,7 @@ def validate_config_file_data(test_config_data: dict):
         return False
 
     if test_config_data.get('error', False):
-        logger.warning(
-            f'Error Occurred While reading file: {test_config_data}')
+        logger.warning('Error Occurred While reading file: %s', test_config_data)
         return False
 
     if not test_config_data.get('actors', ):

src/offat/http.py

@@ -1,7 +1,6 @@
-from aiohttp import ClientSession, ClientTimeout, TCPConnector
+from aiohttp import ClientSession, ClientTimeout
 from aiolimiter import AsyncLimiter
 from os import name as os_name
-from typing import Optional
 
 
 import asyncio

src/offat/logger.py

@@ -14,4 +14,4 @@
         console=console, rich_tracebacks=True, tracebacks_show_locals=True)],
 )
 logger = logging.getLogger("OWASP-OFFAT")
-logger.setLevel(logging.DEBUG)
+logger.setLevel(logging.INFO)

src/offat/openapi.py

@@ -1,9 +1,12 @@
+'''
+module to parse openapi documentation JSON/YAML files.
+'''
 from prance import ResolvingParser
 from .logger import logger
 
 
 class OpenAPIParser:
-    ''''''
+    '''openapi parser'''
 
     def __init__(self, fpath_or_url: str, spec: dict = None) -> None:
         self._parser = ResolvingParser(
@@ -15,35 +18,53 @@ def __init__(self, fpath_or_url: str, spec: dict = None) -> None:
             logger.error('Specification file is invalid!')
 
         self._spec = self._parser.specification
+        self._oas_version = self._get_oas_version()
 
         self.hosts = []
         self._populate_hosts()
         self.host = self.hosts[0]
 
-        self.http_scheme = 'https' if 'https' in self._spec.get(
-            'schemes', []) else 'http'
+        self.http_scheme = self._get_scheme()
         self.api_base_path = self._spec.get('basePath', '')
         self.base_url = f"{self.http_scheme}://{self.host}"
         self.request_response_params = self._get_request_response_params()
 
+    def _get_oas_version(self):
+        if self._spec.get('openapi'):
+            return 3
+        return 2
+
     def _populate_hosts(self):
-        if self._spec.get('openapi'):  # for openapi v3
+        if self._oas_version == 3:
             servers = self._spec.get('servers', [])
             hosts = []
             for server in servers:
                 host = server.get('url', '').removeprefix(
-                    'http://').removeprefix('http://').removesuffix('/')
+                    'https://').removeprefix('http://').removesuffix('/')
                 host = None if host == '' else host
                 hosts.append(host)
         else:
-            host = self._spec.get('host')  # for swagger files
+            host = self._spec.get('host')
             if not host:
                 logger.error('Invalid Host: Host is missing')
                 raise ValueError('Host Not Found in spec file')
             hosts = [host]
 
         self.hosts = hosts
 
+    def _get_scheme(self):
+        if self._oas_version == 3:
+            servers = self._spec.get('servers', [])
+            schemes = []
+            for server in servers:
+                schemes.append('https' if 'https://' in server.get('url', '') else 'http')
+
+            scheme = 'https' if 'https' in schemes else 'http'
+        else:
+            scheme = 'https' if 'https' in self._spec.get('schemes', []) else 'http'
+
+        return scheme
+
     def _get_endpoints(self):
         '''Returns list of endpoint paths along with HTTP methods allowed'''
         endpoints = []

src/offat/report/generator.py

@@ -53,8 +53,7 @@ def handle_report_format(results: list[dict], report_format: str | None) -> str
                 logger.warning('HTML output format displays only basic data.')
                 result = ReportGenerator.generate_html_report(results=results)
             case 'yaml':
-                logger.warning(
-                    'YAML output format needs to be sanitized before using it further.')
+                logger.warning('YAML output format needs to be sanitized before using it further.')
                 result = yaml_dump({
                     'results': results,
                 })
@@ -70,7 +69,7 @@ def handle_report_format(results: list[dict], report_format: str | None) -> str
                     deepcopy(results))
                 result = results_table
 
-        logger.info(f'Generated {report_format.upper()} format report.')
+        logger.info('Generated %s format report.', report_format.upper())
         return result
 
     @staticmethod
@@ -83,7 +82,7 @@ def save_report(report_path: str | None, report_file_content: str | Table | None
         # print to cli if report path and file content as absent else write to file location.
         if report_path and report_file_content and not isinstance(report_file_content, Table):
             with open(report_path, 'w') as f:
-                logger.info(f'Writing report to file: {report_path}')
+                logger.info('Writing report to file: %s', report_path)
                 f.write(report_file_content)
         else:
             if isinstance(report_file_content, Table) and report_file_content.columns:

src/offat/tester/fuzzer.py

@@ -85,6 +85,11 @@ def fill_params(params: list[dict]):
         param_is_required = params[index].get('required')
         param_in = params[index].get('in')
         param_name = params[index].get('name', '')
+        # for OAS 3
+        is_oas_v3 = False
+        if not param_type:
+            is_oas_v3 = True
+            param_type = params[index].get('schema', {}).get('type')
 
         match param_type:
             case 'string':
@@ -94,12 +99,12 @@ def fill_params(params: list[dict]):
             case 'integer':
                 param_value = generate_random_int()
 
-            # TODO: handle file type
+            # TODO: handle file and array type
 
             case _:  # default case
                 param_value = generate_random_string(10)
 
-        if params[index].get('schema'):
+        if params[index].get('schema') and not is_oas_v3:
             schema_obj = params[index].get('schema', {}).get('properties', {})
             filled_schema_params = fill_schema_params(
                 schema_obj, param_in, param_is_required)

src/offat/tester/test_generator.py

@@ -131,8 +131,6 @@ def __get_request_params_list(self, request_params: list[dict]):
                 required_params: list = param_schema.get('required', [])
 
                 for prop in props.keys():
-                    # TODO: handle arrays differently to
-                    # extract their internal params
                     prop_type = props[prop].get('type')
                     payload_data.append({
                         'in': param_pos,
@@ -183,6 +181,11 @@ def __fuzz_request_params(self, openapi_parser: OpenAPIParser) -> list[dict]:
             for path_param in path_params:
                 path_param_name = path_param.get('name')
                 path_param_value = path_param.get('value')
+
+                # below code is for handling OAS 3
+                if not path_param_value:
+                    pass
+
                 endpoint_path = endpoint_path.replace(
                     '{' + str(path_param_name) + '}', str(path_param_value))
 

src/offat/tester/test_runner.py

@@ -1,12 +1,12 @@
 from asyncio import ensure_future, gather
 from enum import Enum
-from rich.progress import Progress, TaskID
+from sys import exc_info
 from traceback import print_exc
+from rich.progress import Progress, TaskID
 
 
 from ..http import AsyncRequests
-from ..logger import logger
-from ..logger import console
+from ..logger import logger, console
 
 
 class PayloadFor(Enum):
@@ -15,9 +15,9 @@ class PayloadFor(Enum):
 
 
 class TestRunner:
-    def __init__(self, rate_limit: float = 60, headers: dict | None = None, proxy: str | None = None, ssl: bool = True) -> None:
+    def __init__(self, rate_limit: float = 60, headers: dict | None = None, proxy: str | None = None) -> None:
         self._client = AsyncRequests(
-            rate_limit=rate_limit, headers=headers, proxy=proxy, ssl=ssl)
+            rate_limit=rate_limit, headers=headers, proxy=proxy)
         self.progress = Progress(console=console)
         self.progress_task_id: TaskID | None = None
 
@@ -43,6 +43,7 @@ def _generate_payloads(self, params: list[dict], payload_for: PayloadFor = Paylo
         query_payload = {}
 
         for param in params:
+
             param_in = param.get('in')
             param_name = param.get('name')
             param_value = param.get('value')
@@ -104,7 +105,8 @@ async def send_request(self, test_task):
             test_result['redirection'] = ''
             test_result['error'] = True
 
-            logger.error(f'Unable to send request due to error: {e}')
+            logger.debug('Exception Debug Data:', exc_info=exc_info())
+            logger.error('Unable to send request due to error: %s', e)
             logger.error(locals())
 
         # advance progress bar

src/offat/tester/tester_utils.py

@@ -8,7 +8,6 @@
 from .test_runner import TestRunner
 from ..report.generator import ReportGenerator
 from ..logger import logger
-from ..http import AsyncRequests
 from ..openapi import OpenAPIParser
 
 
@@ -17,6 +16,8 @@
 
 
 def is_host_up(openapi_parser: OpenAPIParser) -> bool:
+    '''checks whether the host from openapi doc is available or not. 
+    Returns True is host is available else returns False'''
     tokens = openapi_parser.host.split(":")
     match len(tokens):
         case 1:
@@ -26,24 +27,33 @@ def is_host_up(openapi_parser: OpenAPIParser) -> bool:
             host = tokens[0]
             port = tokens[1]
         case _:
-            logger.warning(f"Invalid host: {openapi_parser.host}")
+            logger.warning("Invalid host: %s", openapi_parser.host)
             return False
 
-    logger.info(f"Checking whether host {host}:{port} is available")
+    host = host.split('/')[0]
+
+    match port:
+        case 443:
+            proto = http_client.HTTPSConnection
+        case _:
+            proto = http_client.HTTPConnection
+
+    logger.info("Checking whether host %s:%d is available", host, port)
     try:
-        conn = http_client.HTTPConnection(host=host, port=port, timeout=5)
+        conn = proto(host=host, port=port, timeout=5)
         conn.request("GET", "/")
         res = conn.getresponse()
-        logger.info(f"Host returned status code: {res.status}")
+        logger.info("Host returned status code: %d", res.status)
         return res.status in range(200, 499)
     except Exception as e:
-        logger.error(
-            f"Unable to connect to host {host}:{port} due to error: {e}")
+        logger.error("Unable to connect to host %s:%d due to error: %s", host, port, repr(e))
         return False
 
 
 def run_test(test_runner: TestRunner, tests: list[dict], regex_pattern: Optional[str] = None, skip_test_run: Optional[bool] = False, post_run_matcher_test: Optional[bool] = False, description: Optional[str] = None) -> list:
     '''Run tests and print result on console'''
+    logger.info('Tests Generated: %d', len(tests))
+
     # filter data if regex is passed
     if regex_pattern:
         tests = list(
@@ -75,20 +85,19 @@ def run_test(test_runner: TestRunner, tests: list[dict], regex_pattern: Optional
 
 
 # Note: redirects are allowed by default making it easier for pentesters/researchers
-def generate_and_run_tests(api_parser: OpenAPIParser, regex_pattern: Optional[str] = None, output_file: Optional[str] = None, output_file_format: Optional[str] = None, rate_limit: Optional[int] = None, delay: Optional[float] = None, req_headers: Optional[dict] = None, proxy: Optional[str] = None, ssl: Optional[bool] = True, test_data_config: Optional[dict] = None):
+def generate_and_run_tests(api_parser: OpenAPIParser, regex_pattern: Optional[str] = None, output_file: Optional[str] = None, output_file_format: Optional[str] = None, rate_limit: Optional[int] = None, req_headers: Optional[dict] = None, proxy: Optional[str] = None, test_data_config: Optional[dict] = None):
     global test_table_generator, logger
 
     if not is_host_up(openapi_parser=api_parser):
-        logger.error(
-            f"Stopping tests due to unavailibility of host: {api_parser.host}")
+        logger.error("Stopping tests due to unavailibility of host: %s", api_parser.host)
         return
-    logger.info(f"Host {api_parser.host} is up")
+
+    logger.info("Host %s is up", api_parser.host)
 
     test_runner = TestRunner(
         rate_limit=rate_limit,
         headers=req_headers,
         proxy=proxy,
-        ssl=ssl,
     )
 
     results: list = []