Merge pull request #42 from OWASP/dev

Dev RELEASE

README.md

@@ -6,41 +6,41 @@ OWASP OFFAT (OFFensive Api Tester) is created to automatically test API for comm
 
 ## Demo
 
-[![ASCII Cast Demo](https://asciinema.org/a/9MSwl7UafIVT3iJn13OcvWXeF.svg)](https://asciinema.org/a/9MSwl7UafIVT3iJn13OcvWXeF)
+[![asciicast](https://asciinema.org/a/LFXLILNkf7Gce5uCuJydplbEd.svg)](https://asciinema.org/a/LFXLILNkf7Gce5uCuJydplbEd)
 
 ## Security Checks
 
-- Restricted HTTP Methods
-- SQLi
-- BOLA
-- Data Exposure
-- BOPLA / Mass Assignment
-- Broken Access Control
-- Basic Command Injection
-- Basic XSS/HTML Injection test
+-   Restricted HTTP Methods
+-   SQLi
+-   BOLA
+-   Data Exposure
+-   BOPLA / Mass Assignment
+-   Broken Access Control
+-   Basic Command Injection
+-   Basic XSS/HTML Injection test
 
 ## Features
 
-- Few Security Checks from OWASP API Top 10
-- Automated Testing
-- User Config Based Testing
-- API for Automating tests and Integrating Tool with other platforms/tools
-- CLI tool
-- Dockerized Project for Easy Usage
-- Open Source Tool with MIT License
+-   Few Security Checks from OWASP API Top 10
+-   Automated Testing
+-   User Config Based Testing
+-   API for Automating tests and Integrating Tool with other platforms/tools
+-   CLI tool
+-   Dockerized Project for Easy Usage
+-   Open Source Tool with MIT License
 
 ## Try Tool
 
-- Install Tool using pip
+-   Install Tool using pip
 
 ```bash
 python -m pip install offat
 ```
 
-- Run Tool
+-   Run Tool
 
 ```bash
 offat -f swagger_file.json
 ```
 
-- For more usage options read [README.md](https://github.com/OWASP/OFFAT/blob/main/src/README.md)
+-   For more usage options read [README.md](https://github.com/OWASP/OFFAT/blob/main/src/README.md)

src/offat/http.py

@@ -50,9 +50,9 @@ async def request(self, url: str, method: str = 'GET', session: ClientSession =
             dict: returns request and response data as dict
         '''
         is_new_session = False
-        connector = TCPConnector(ssl=self._ssl, limit=self._rate_limit,)
 
         if not session:
+            connector = TCPConnector(ssl=self._ssl, limit=self._rate_limit,)
             session = ClientSession(headers=self._headers, connector=connector)
             is_new_session = True
 

src/offat/tester/test_runner.py

@@ -91,6 +91,7 @@ async def send_request(self, test_task):
             logger.error('Connection Failed! Server refused Connection!!')
         except ClientProxyConnectionError as e:
             logger.error(f'Proxy Connection Error: {e}')
+        # TODO: handle exception here
 
         test_result = test_task
 

src/offat/tester/tester_utils.py

@@ -1,5 +1,5 @@
 from asyncio import run
-from copy import deepcopy
+from http import client as http_client
 from typing import Optional
 from re import search as regex_search
 
@@ -8,13 +8,40 @@
 from .test_runner import TestRunner
 from ..report.generator import ReportGenerator
 from ..logger import logger
+from ..http import AsyncRequests
 from ..openapi import OpenAPIParser
 
 
 # create tester objs
 test_generator = TestGenerator()
 
 
+def is_host_up(openapi_parser: OpenAPIParser) -> bool:
+    tokens = openapi_parser.host.split(":")
+    match len(tokens):
+        case 1:
+            host = tokens[0]
+            port = 443 if openapi_parser.http_scheme == "https" else 80
+        case 2:
+            host = tokens[0]
+            port = tokens[1]
+        case _:
+            logger.warning(f"Invalid host: {openapi_parser.host}")
+            return False
+
+    logger.info(f"Checking whether host {host}:{port} is available")
+    try:
+        conn = http_client.HTTPConnection(host=host, port=port, timeout=5)
+        conn.request("GET", "/")
+        res = conn.getresponse()
+        logger.info(f"Host returned status code: {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}")
+        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'''
     # filter data if regex is passed
@@ -51,6 +78,12 @@ def run_test(test_runner: TestRunner, tests: list[dict], regex_pattern: Optional
 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):
     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}")
+        return
+    logger.info(f"Host {api_parser.host} is up")
+
     test_runner = TestRunner(
         rate_limit=rate_limit,
         delay=delay,
@@ -239,10 +272,17 @@ def generate_and_run_tests(api_parser: OpenAPIParser, regex_pattern: Optional[st
         )
 
     # save file to output if output flag is present
+    if output_file_format != 'table':
+        ReportGenerator.generate_report(
+            results=results,
+            report_format=output_file_format,
+            report_path=output_file,
+        )
+
     ReportGenerator.generate_report(
         results=results,
-        report_format=output_file_format,
-        report_path=output_file,
+        report_format='table',
+        report_path=None,
     )
 
     return results

src/pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "offat"
-version = "0.12.4"
+version = "0.13.0"
 description = "Offensive API tester tool automates checks for common API vulnerabilities"
 authors = ["Dhrumil Mistry <dhrumil.mistry@owasp.org>"]
 license = "MIT"