fix ruff and mypy lint, add pre commit so it doesnt happen again

This commit is contained in:
CarterPerez-dev 2025-11-12 15:30:47 -05:00
parent 0c4a22748c
commit e7a4ea6f92
43 changed files with 591 additions and 1217 deletions

View File

@ -12,24 +12,24 @@ jobs:
lint:
name: Run Linters
runs-on: ubuntu-latest
permissions:
pull-requests: write
contents: read
defaults:
run:
working-directory: PROJECTS/api-security-scanner/backend
steps:
- name: Checkout code
uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: '3.12'
- name: Cache pip dependencies
uses: actions/cache@v4
with:
@ -37,12 +37,12 @@ jobs:
key: ${{ runner.os }}-pip-${{ hashFiles('PROJECTS/api-security-scanner/backend/pyproject.toml') }}
restore-keys: |
${{ runner.os }}-pip-
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install -e ".[dev]"
- name: Run pylint
id: pylint
run: |
@ -56,7 +56,7 @@ jobs:
fi
cat pylint-output.txt
continue-on-error: true
- name: Run ruff
id: ruff
run: |
@ -70,7 +70,7 @@ jobs:
fi
cat ruff-output.txt
continue-on-error: true
- name: Run mypy
id: mypy
run: |
@ -84,7 +84,7 @@ jobs:
fi
cat mypy-output.txt
continue-on-error: true
- name: Create Lint Summary
id: create_summary
if: github.event_name == 'pull_request'
@ -92,7 +92,7 @@ jobs:
{
echo '## 🔍 Lint & Type Check Results'
echo ''
# Pylint Status
if [[ "${{ env.PYLINT_PASSED }}" == "true" ]]; then
echo '### ✅ Pylint: **Passed**'
@ -107,7 +107,7 @@ jobs:
echo '</details>'
fi
echo ''
# Ruff Status
if [[ "${{ env.RUFF_PASSED }}" == "true" ]]; then
echo '### ✅ Ruff: **Passed**'
@ -122,7 +122,7 @@ jobs:
echo '</details>'
fi
echo ''
# Mypy Status
if [[ "${{ env.MYPY_PASSED }}" == "true" ]]; then
echo '### ✅ Mypy: **Passed**'
@ -137,7 +137,7 @@ jobs:
echo '</details>'
fi
echo ''
# Overall Summary
if [[ "${{ env.PYLINT_PASSED }}" == "true" ]] && [[ "${{ env.RUFF_PASSED }}" == "true" ]] && [[ "${{ env.MYPY_PASSED }}" == "true" ]]; then
echo '---'
@ -149,7 +149,7 @@ jobs:
echo ''
echo '<!-- lint-check-comment-marker -->'
} > ../lint-report.md
- name: Post PR Comment
if: github.event_name == 'pull_request'
uses: peter-evans/create-or-update-comment@v4

62
.pre-commit-config.yaml Normal file
View File

@ -0,0 +1,62 @@
repos:
# Python Backend Checks
- repo: https://github.com/astral-sh/ruff-pre-commit
rev: v0.8.4
hooks:
- id: ruff
name: ruff check (backend)
args: [--fix, --exit-non-zero-on-fix]
files: ^PROJECTS/api-security-scanner/backend/
exclude: ^PROJECTS/api-security-scanner/backend/(\.venv|__pycache__|\.pytest_cache)/
- id: ruff-format
name: ruff format (backend)
files: ^PROJECTS/api-security-scanner/backend/
exclude: ^PROJECTS/api-security-scanner/backend/(\.venv|__pycache__|\.pytest_cache)/
- repo: local
hooks:
- id: mypy
name: mypy type check (backend)
entry: bash -c 'cd PROJECTS/api-security-scanner/backend && mypy .'
language: system
types: [python]
files: ^PROJECTS/api-security-scanner/backend/
pass_filenames: false
- id: pylint
name: pylint check (backend)
entry: bash -c 'cd PROJECTS/api-security-scanner/backend && pylint **/*.py'
language: system
types: [python]
files: ^PROJECTS/api-security-scanner/backend/
pass_filenames: false
# Frontend TypeScript/ESLint Checks
- id: eslint
name: eslint check (frontend)
entry: bash -c 'cd PROJECTS/api-security-scanner/frontend && npm run lint:eslint'
language: system
types_or: [ts, tsx, javascript, jsx]
files: ^PROJECTS/api-security-scanner/frontend/src/
pass_filenames: false
- id: typescript
name: TypeScript type check (frontend)
entry: bash -c 'cd PROJECTS/api-security-scanner/frontend && npm run lint:types'
language: system
types_or: [ts, tsx]
files: ^PROJECTS/api-security-scanner/frontend/src/
pass_filenames: false
# General File Checks
- repo: https://github.com/pre-commit/pre-commit-hooks
rev: v5.0.0
hooks:
- id: trailing-whitespace
- id: end-of-file-fixer
- id: check-yaml
- id: check-json
- id: check-merge-conflict
- id: check-added-large-files
args: ['--maxkb=1000']

View File

@ -1,120 +0,0 @@
name: ESLint Check
on:
pull_request:
branches: [ '*' ]
paths:
- 'frontend/**/*.ts'
- 'frontend/**/*.tsx'
- 'frontend/eslint.config.js'
- 'frontend/tsconfig*.json'
- 'frontend/package.json'
- '.github/workflows/eslint-check.yml'
jobs:
eslint-check:
name: ESLint TypeScript Check
runs-on: ubuntu-latest
permissions:
pull-requests: write
contents: read
defaults:
run:
working-directory: frontend
steps:
- name: Checkout code
uses: actions/checkout@v4
- name: Setup Node.js
uses: actions/setup-node@v4
with:
node-version: '20'
cache: 'npm'
cache-dependency-path: frontend/package-lock.json
- name: Install dependencies
run: npm install
- name: Run ESLint
id: eslint_check
run: |
echo "Running ESLint on TypeScript/React files..."
if npm run lint:eslint > eslint-output.txt 2>&1; then
echo "ESLINT_PASSED=true" >> $GITHUB_ENV
echo "✅ No ESLint errors found!"
echo "ERROR_COUNT=0" >> $GITHUB_ENV
else
echo "ESLINT_PASSED=false" >> $GITHUB_ENV
# Count error lines (lines that contain file paths with problems)
error_count=$(grep -c "^/" eslint-output.txt || echo "0")
echo "ERROR_COUNT=$error_count" >> $GITHUB_ENV
echo "⚠️ ESLint found issues in $error_count files!"
fi
cat eslint-output.txt
continue-on-error: true
- name: Create ESLint Summary
id: create_summary
if: github.event_name == 'pull_request'
run: |
{
echo '## 🔍 ESLint Results'
echo ''
if [[ "${{ env.ESLINT_PASSED }}" == "true" ]]; then
echo '### ✅ **Perfect! No ESLint issues found** 🎉'
echo ''
echo 'Your TypeScript and React code follows all the coding standards perfectly!'
echo ''
echo '**What was checked:**'
echo '- TypeScript strict type checking and stylistic rules'
echo '- React component patterns and hooks usage'
echo '- Code complexity, naming conventions, and best practices'
echo '- Accessibility (jsx-a11y) and React Refresh compatibility'
else
echo '### ❌ **ESLint found issues in ${{ env.ERROR_COUNT }} files**'
echo ''
echo 'Please review and fix the TypeScript/React issues below:'
echo ''
echo '<details><summary>📋 View detailed ESLint output</summary>'
echo ''
echo '```'
head -100 eslint-output.txt
echo '```'
echo '</details>'
echo ''
echo '**How to fix:**'
echo '1. Run `cd frontend && npm run lint:eslint` locally to see the issues'
echo '2. Fix the reported TypeScript, React, and code quality problems'
echo '3. For auto-fixable issues: `cd frontend && npx eslint . --ext .ts,.tsx --fix`'
echo '4. Push your changes to update this PR'
fi
echo ''
echo '**Commands:**'
echo '- `cd frontend && npm run lint:eslint` - Run ESLint'
echo '- `cd frontend && npx eslint . --ext .ts,.tsx --fix` - Auto-fix issues'
echo '- ESLint config: `frontend/eslint.config.js`'
echo ''
echo '<!-- eslint-check-comment-marker -->'
} > eslint-report.md
- name: Post PR Comment
if: github.event_name == 'pull_request'
uses: peter-evans/create-or-update-comment@v4
with:
issue-number: ${{ github.event.pull_request.number }}
body-path: frontend/eslint-report.md
edit-mode: replace
- name: Exit with proper code
run: |
if [[ "${{ env.ESLINT_PASSED }}" == "false" ]]; then
echo "❌ ESLint checks failed. Please fix the issues above."
exit 1
else
echo "✅ All ESLint checks passed!"
exit 0
fi

View File

@ -1,159 +0,0 @@
name: Lint & Type Check
on:
push:
branches: [ 'main' ]
pull_request:
branches: [ '*' ]
workflow_dispatch:
jobs:
lint:
name: Run Linters
runs-on: ubuntu-latest
permissions:
pull-requests: write
contents: read
defaults:
run:
working-directory: backend
steps:
- name: Checkout code
uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: '3.12'
- name: Cache pip dependencies
uses: actions/cache@v4
with:
path: ~/.cache/pip
key: ${{ runner.os }}-pip-${{ hashFiles('backend/pyproject.toml') }}
restore-keys: |
${{ runner.os }}-pip-
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install -e ".[dev]"
- name: Run pylint
id: pylint
run: |
echo "Running pylint..."
if pylint . > pylint-output.txt 2>&1; then
echo "PYLINT_PASSED=true" >> $GITHUB_ENV
echo "✅ No pylint errors found!"
else
echo "PYLINT_PASSED=false" >> $GITHUB_ENV
echo "⚠️ Pylint found issues!"
fi
cat pylint-output.txt
continue-on-error: true
- name: Run ruff
id: ruff
run: |
echo "Running ruff check..."
if ruff check . > ruff-output.txt 2>&1; then
echo "RUFF_PASSED=true" >> $GITHUB_ENV
echo "✅ No ruff errors found!"
else
echo "RUFF_PASSED=false" >> $GITHUB_ENV
echo "⚠️ Ruff found issues"
fi
cat ruff-output.txt
continue-on-error: true
- name: Run mypy
id: mypy
run: |
echo "Running mypy..."
if mypy . > mypy-output.txt 2>&1; then
echo "MYPY_PASSED=true" >> $GITHUB_ENV
echo "✅ No mypy errors found"
else
echo "MYPY_PASSED=false" >> $GITHUB_ENV
echo "⚠️ Mypy found issues"
fi
cat mypy-output.txt
continue-on-error: true
- name: Create Lint Summary
id: create_summary
if: github.event_name == 'pull_request'
run: |
{
echo '## 🔍 Lint & Type Check Results'
echo ''
# Pylint Status
if [[ "${{ env.PYLINT_PASSED }}" == "true" ]]; then
echo '### ✅ Pylint: **Passed**'
echo 'No pylint issues found.'
else
echo '### ⚠️ Pylint: **Issues Found**'
echo '<details><summary>View pylint output</summary>'
echo ''
echo '```'
head -100 pylint-output.txt
echo '```'
echo '</details>'
fi
echo ''
# Ruff Status
if [[ "${{ env.RUFF_PASSED }}" == "true" ]]; then
echo '### ✅ Ruff: **Passed**'
echo 'No ruff issues found.'
else
echo '### ⚠️ Ruff: **Issues Found**'
echo '<details><summary>View ruff output</summary>'
echo ''
echo '```'
head -100 ruff-output.txt
echo '```'
echo '</details>'
fi
echo ''
# Mypy Status
if [[ "${{ env.MYPY_PASSED }}" == "true" ]]; then
echo '### ✅ Mypy: **Passed**'
echo 'No mypy issues found.'
else
echo '### ⚠️ Mypy: **Issues Found**'
echo '<details><summary>View mypy output</summary>'
echo ''
echo '```'
head -100 mypy-output.txt
echo '```'
echo '</details>'
fi
echo ''
# Overall Summary
if [[ "${{ env.PYLINT_PASSED }}" == "true" ]] && [[ "${{ env.RUFF_PASSED }}" == "true" ]] && [[ "${{ env.MYPY_PASSED }}" == "true" ]]; then
echo '---'
echo '### All checks passed!'
else
echo '---'
echo '### Review the issues above and consider fixing them.'
fi
echo ''
echo '<!-- lint-check-comment-marker -->'
} > ../lint-report.md
- name: Post PR Comment
if: github.event_name == 'pull_request'
uses: peter-evans/create-or-update-comment@v4
with:
issue-number: ${{ github.event.pull_request.number }}
body-path: lint-report.md
comment-marker: lint-check-comment-marker

View File

@ -1,129 +0,0 @@
name: TypeScript Type Check
on:
pull_request:
branches: [ '*' ]
paths:
- 'frontend/**/*.ts'
- 'frontend/**/*.tsx'
- 'frontend/tsconfig*.json'
- 'frontend/package.json'
- '.github/workflows/typescript-check.yml'
jobs:
typescript-check:
name: TypeScript Type Check
runs-on: ubuntu-latest
permissions:
pull-requests: write
contents: read
defaults:
run:
working-directory: frontend
steps:
- name: Checkout code
uses: actions/checkout@v4
- name: Setup Node.js
uses: actions/setup-node@v4
with:
node-version: '20'
cache: 'npm'
cache-dependency-path: frontend/package-lock.json
- name: Install dependencies
run: |
for i in {1..3}; do
echo "Attempt $i of 3..."
if npm install; then
echo "✅ npm install succeeded"
break
else
echo "⚠️ npm install failed, retrying in 10 seconds..."
sleep 10
fi
done
- name: Run TypeScript type checking
id: typescript_check
run: |
echo "Running TypeScript type checking..."
if npm run lint:types > typescript-output.txt 2>&1; then
echo "TYPESCRIPT_PASSED=true" >> $GITHUB_ENV
echo "✅ No TypeScript type errors found!"
echo "ERROR_COUNT=0" >> $GITHUB_ENV
else
echo "TYPESCRIPT_PASSED=false" >> $GITHUB_ENV
# Count error lines (lines that contain errors)
error_count=$(grep -c "error TS" typescript-output.txt || echo "0")
echo "ERROR_COUNT=$error_count" >> $GITHUB_ENV
echo "⚠️ TypeScript found $error_count type errors!"
fi
cat typescript-output.txt
continue-on-error: true
- name: Create TypeScript Summary
id: create_summary
if: github.event_name == 'pull_request'
run: |
{
echo '## 📝 TypeScript Type Check Results'
echo ''
if [[ "${{ env.TYPESCRIPT_PASSED }}" == "true" ]]; then
echo '### ✅ **Perfect! No TypeScript type errors found** 🎉'
echo ''
echo 'Your TypeScript code passes all strict type checking requirements!'
echo ''
echo '**What was checked:**'
echo '- Strict type checking with `exactOptionalPropertyTypes`'
echo '- No unused locals or parameters'
echo '- Proper return types and void expressions'
echo '- Module resolution and import/export syntax'
else
echo '### ❌ **TypeScript found ${{ env.ERROR_COUNT }} type errors**'
echo ''
echo 'Please review and fix the TypeScript type errors below:'
echo ''
echo '<details><summary>📋 View detailed TypeScript output</summary>'
echo ''
echo '```'
head -100 typescript-output.txt
echo '```'
echo '</details>'
echo ''
echo '**How to fix:**'
echo '1. Run `cd frontend && npm run lint:types` locally to see the type errors'
echo '2. Fix the reported TypeScript type issues'
echo '3. Ensure all variables have proper types and return types are explicit'
echo '4. Push your changes to update this PR'
fi
echo ''
echo '**Commands:**'
echo '- `cd frontend && npm run lint:types` - Run TypeScript type checking'
echo '- `cd frontend && npm run build` - Run full build with type checking'
echo '- TypeScript config: `frontend/tsconfig.app.json`'
echo ''
echo '<!-- typescript-check-comment-marker -->'
} > typescript-report.md
- name: Post PR Comment
if: github.event_name == 'pull_request'
uses: peter-evans/create-or-update-comment@v4
with:
issue-number: ${{ github.event.pull_request.number }}
body-path: frontend/typescript-report.md
edit-mode: replace
- name: Exit with proper code
run: |
if [[ "${{ env.TYPESCRIPT_PASSED }}" == "false" ]]; then
echo "❌ TypeScript type checking failed. Please fix the type errors above."
exit 1
else
echo "✅ All TypeScript type checks passed!"
exit 0
fi

View File

@ -1,5 +1,5 @@
# ⒸAngelaMos | 2025
# Gitignore - This is where you list files/folders that
# Gitignore - This is where you list files/folders that
# should not be publicly commited to the remote repository
# Python

View File

@ -41,17 +41,17 @@ help:
@echo "\033[1;4;91mSyntax Checking (Ruff):\033[0m"
@echo " \033[92mmake \033[36mqa\033[0m \033[94m- Check all code for syntax issues\033[0m"
@echo "\033[1;4;91mSyntax Fixing (Ruff):\033[0m"
@echo " \033[92mmake \033[36mqa\033[0m \033[94m- Fix all code of syntax issues\033[0m"
@echo " \033[92mmake \033[36mqa\033[0m \033[94m- Fix all code of syntax issues\033[0m"
@echo ""
@echo "\033[1;4;94mType Checking (mypy):\033[0m"
@echo " \033[92mmake \033[36mqa\033[0m \033[94m- Type check all code\033[0m"
@echo "\033[1;4;94mFormat:\033[0m"
@echo " \033[92mmake \033[36mqa\033[0m \033[94m- Format code\033[0m"
@echo " \033[92mmake \033[36mqa\033[0m \033[94m- Format code\033[0m"
@echo ""
@echo "\033[1;4;92mUtilities:\033[0m"
@echo " \033[92mmake \033[36mclean\033[0m \033[94m- Remove all cache files\033[0m"
@echo " \033[92mmake \033[36mtree\033[0m \033[94m- Display the file tree\033[0m"
@echo " \033[92mmake \033[36mTODO\033[0m \033[94m- Find All TODO \033[0m"
@echo " \033[92mmake \033[36mTODO\033[0m \033[94m- Find All TODO \033[0m"
@echo ""
@echo "\033[1;93mCurrent default domain: \033[1;5;95m$(DOMAIN)\033[0m"
@echo ""
@ -59,7 +59,7 @@ help:
@echo " \033[91maccount\033[0m, \033[92mcommunity\033[0m, \033[93mcontent\033[0m, \033[94mfreemium\033[0m, \033[95mgames\033[0m,"
@echo " \033[96mmarketing\033[0m, \033[91mprogression\033[0m, \033[92mshop\033[0m, \033[93mtesting\033[0m, \033[94mtools\033[0m"
@echo ""
@echo "\033[95m ------------------------------"
@echo "\033[95m ------------------------------"
@echo "\033[34m ⣿⣿⣿⡷⠊⡢⡹⣦⡑⢂⢕⢂⢕⢂⢕⢂⠕⠔⠌⠝⠛⠶⠶⢶⣦⣄⢂⢕⢂⢕"
@echo "\033[34m ⣿⣿⠏⣠⣾⣦⡐⢌⢿⣷⣦⣅⡑⠕⠡⠐⢿⠿⣛⠟⠛⠛⠛⠛⠡⢷⡈⢂⢕⢂"
@echo "\033[34m ⠟⣡⣾⣿⣿⣿⣿⣦⣑⠝⢿⣿⣿⣿⣿⣿⡵⢁⣤⣶⣶⣿⢿⢿⢿⡟⢻⣤⢑⢂"
@ -70,7 +70,7 @@ help:
@echo "\033[32m ⠣⡁⠹⡪⡪⡪⡪⣪⣾⣿⣿⣿⣿⠋⠐⢉⢍⢄⢌⠻⣿⣿⣿⣿⣿⣿⣿⣿⠏⠈"
@echo "\033[32m ⡣⡘⢄⠙⣾⣾⣾⣿⣿⣿⣿⣿⣿⡀⢐⢕⢕⢕⢕⢕⡘⣿⣿⣿⣿⣿⣿⠏⠠⠈"
@echo "\033[32m ⠌⢊⢂⢣⠹⣿⣿⣿⣿⣿⣿⣿⣿⣧⢐⢕⢕⢕⢕⢕⢅⣿⣿⣿⣿⡿⢋⢜⠠⠈"
@echo "\033[95m ------------------------------"
@echo "\033[95m ------------------------------"
@echo "\033[0m"
@ -104,20 +104,20 @@ check:
@ruff check .
$(call ASCII)
# ==================== Syntax Checking (Ruff) ====================
fix:
@echo "\033[1;3;4;96m======== ruff fix All Code ========\033[0m"
@ruff check . --fix
$(call ASCII)
# ======================= In Depth Linting (pylint)=======================
pylint:
@echo "\033[1;3;4;96m======== Linting All Code ========\033[0m"
@pylint .
$(call ASCII)
# ======================= Type Checking (mypy) =======================
mypy:
@echo "\033[1;3;4;96m======== Type Checking All Code ========\033[0m"
@ -146,9 +146,8 @@ tree:
@echo "\033[1;3;4;96m======== Creating Tree ========\033[0m"
@tree -I 'node_modules|.git|*.log|dist|build|*.cache|certgames.egg-info|context|.venv|*.env|qa_context'
$(call ASCII)
TODO:
@echo "\033[1;3;4;96m======== Looking for TODO's ========\033[0m"
@find . -name "*.py" -print0 | xargs -0 pylint --disable=all --enable=W0511 --msg-template='{path}:{line}:{column}: {msg_id}: {msg} ({symbol})' || true
$(call ASCII)

View File

@ -1,5 +1,5 @@
"""
AngelaMos | CarterPerez-dev
AngelaMos | CarterPerez-dev
CertGames.com | 2025
----
API Security Scanner
@ -15,6 +15,6 @@ API Security Scanner
"""

View File

@ -16,9 +16,7 @@ class Settings(BaseSettings):
"""
model_config = SettingsConfigDict(
env_file = "../.env",
env_file_encoding = "utf-8",
case_sensitive = True
env_file="../.env", env_file_encoding="utf-8", case_sensitive=True
)
# Application metadata
@ -86,9 +84,7 @@ class Settings(BaseSettings):
"""
Convert comma separated CORS origins string to list
"""
return [
origin.strip() for origin in self.CORS_ORIGINS.split(",")
]
return [origin.strip() for origin in self.CORS_ORIGINS.split(",")]
@lru_cache

View File

@ -13,16 +13,12 @@ from config import settings
# Database engine
engine = create_engine(
settings.DATABASE_URL,
pool_pre_ping = True,
echo = settings.DEBUG,
pool_pre_ping=True,
echo=settings.DEBUG,
)
# Session factory
SessionLocal = sessionmaker(
autocommit = False,
autoflush = False,
bind = engine
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
# Base class
Base = declarative_base()

View File

@ -34,25 +34,25 @@ async def get_current_user(
if email is None:
raise HTTPException(
status_code = status.HTTP_401_UNAUTHORIZED,
detail = "Invalid authentication credentials",
headers = {"WWW-Authenticate": "Bearer"},
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid authentication credentials",
headers={"WWW-Authenticate": "Bearer"},
)
user = UserRepository.get_by_email(db, email)
if not user:
raise HTTPException(
status_code = status.HTTP_401_UNAUTHORIZED,
detail = "User not found",
headers = {"WWW-Authenticate": "Bearer"},
status_code=status.HTTP_401_UNAUTHORIZED,
detail="User not found",
headers={"WWW-Authenticate": "Bearer"},
)
return UserResponse.model_validate(user)
except ValueError:
raise HTTPException(
status_code = status.HTTP_401_UNAUTHORIZED,
detail = "Invalid authentication credentials",
headers = {"WWW-Authenticate": "Bearer"},
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid authentication credentials",
headers={"WWW-Authenticate": "Bearer"},
) from None

View File

@ -9,6 +9,7 @@ class ScanStatus(str, Enum):
"""
Enum for scan result status
"""
VULNERABLE = "vulnerable"
SAFE = "safe"
ERROR = "error"
@ -18,6 +19,7 @@ class Severity(str, Enum):
"""
Enum for vulnerability severity levels
"""
CRITICAL = "critical"
HIGH = "high"
MEDIUM = "medium"
@ -29,6 +31,7 @@ class TestType(str, Enum):
"""
Enum for available security test types
"""
RATE_LIMIT = "rate_limit"
AUTH = "auth"
SQLI = "sqli"

View File

@ -16,29 +16,22 @@ def hash_password(password: str) -> str:
"""
Hash a plain text password using bcrypt
"""
password_bytes = password.encode('utf-8')
password_bytes = password.encode("utf-8")
salt = bcrypt.gensalt()
hashed = bcrypt.hashpw(password_bytes, salt)
return hashed.decode('utf-8')
return hashed.decode("utf-8")
def verify_password(
plain_password: str,
hashed_password: str
) -> bool:
def verify_password(plain_password: str, hashed_password: str) -> bool:
"""
Verify a plain text password against a hashed password
"""
password_bytes = plain_password.encode('utf-8')
hashed_bytes = hashed_password.encode('utf-8')
password_bytes = plain_password.encode("utf-8")
hashed_bytes = hashed_password.encode("utf-8")
return bcrypt.checkpw(password_bytes, hashed_bytes)
def create_access_token(
data: dict[str,
str],
expires_delta: timedelta | None = None
) -> str:
def create_access_token(data: dict[str, str], expires_delta: timedelta | None = None) -> str:
"""
Create a JWT access token
"""
@ -47,16 +40,10 @@ def create_access_token(
if expires_delta:
expire = datetime.utcnow() + expires_delta
else:
expire = datetime.utcnow() + timedelta(
minutes = settings.ACCESS_TOKEN_EXPIRE_MINUTES
)
expire = datetime.utcnow() + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
to_encode.update({"exp": expire})
encoded_jwt = jwt.encode(
to_encode,
settings.SECRET_KEY,
algorithm = settings.ALGORITHM
)
encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
return encoded_jwt
@ -65,11 +52,7 @@ def decode_token(token: str) -> dict[str, str]:
Decode and verify a JWT token
"""
try:
payload = jwt.decode(
token,
settings.SECRET_KEY,
algorithms = [settings.ALGORITHM]
)
payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])
return payload
except JWTError as e:
raise ValueError(f"Invalid token: {str(e)}") from e

View File

@ -22,31 +22,28 @@ def create_app() -> FastAPI:
"""
Application factory function
"""
Base.metadata.create_all(bind = engine)
Base.metadata.create_all(bind=engine)
app = FastAPI(
title = settings.APP_NAME,
version = settings.VERSION,
openapi_version = "3.1.0",
docs_url = "/docs",
redoc_url = "/redoc",
openapi_url = "/openapi.json",
debug = settings.DEBUG,
title=settings.APP_NAME,
version=settings.VERSION,
openapi_version="3.1.0",
docs_url="/docs",
redoc_url="/redoc",
openapi_url="/openapi.json",
debug=settings.DEBUG,
)
limiter = Limiter(key_func = get_remote_address)
limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
app.add_exception_handler(
RateLimitExceeded,
_rate_limit_exceeded_handler
)
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
app.add_middleware(
CORSMiddleware,
allow_origins = settings.cors_origins_list,
allow_credentials = True,
allow_methods = ["*"],
allow_headers = ["*"],
allow_origins=settings.cors_origins_list,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
_register_routes(app)
@ -58,6 +55,7 @@ def _register_routes(app: FastAPI) -> None:
"""
Register all application routes
"""
@app.get("/")
def root() -> dict[str, str]:
"""

View File

@ -1,6 +1,6 @@
"""
CertGames.com | 2025
AngelaMos | CarterPerez-dev
AngelaMos | CarterPerez-dev
----
API Security Scanner FastAPI entry point
"""

View File

@ -1,6 +1,6 @@
"""
AngelaMos | 2025
Base model class
Base model class
Common fields and methods for all models
"""
@ -21,22 +21,15 @@ class BaseModel(Base):
Abstract base model with common fields and methods
All models inherit from this class
"""
__abstract__ = True
id = Column(
Integer,
primary_key = True,
index = True,
autoincrement = True
)
created_at = Column(
DateTime(timezone = True),
default = lambda: datetime.now(UTC)
)
id = Column(Integer, primary_key=True, index=True, autoincrement=True)
created_at = Column(DateTime(timezone=True), default=lambda: datetime.now(UTC))
updated_at = Column(
DateTime(timezone = True),
default = lambda: datetime.now(UTC),
onupdate = lambda: datetime.now(UTC),
DateTime(timezone=True),
default=lambda: datetime.now(UTC),
onupdate=lambda: datetime.now(UTC),
)
@declared_attr
@ -53,11 +46,7 @@ class BaseModel(Base):
Returns:
dict: Dictionary representation of the model
"""
return {
column.name: getattr(self,
column.name)
for column in self.__table__.columns
}
return {column.name: getattr(self, column.name) for column in self.__table__.columns}
def update(self, **kwargs: Any) -> None:
"""

View File

@ -24,30 +24,30 @@ class Scan(BaseModel):
"""
Stores metadata about scans performed on target URLs
"""
__tablename__ = "scans"
user_id = Column(
Integer,
ForeignKey("users.id",
ondelete = "CASCADE"),
nullable = False,
index = True,
ForeignKey("users.id", ondelete="CASCADE"),
nullable=False,
index=True,
)
target_url = Column(
String(settings.URL_MAX_LENGTH),
nullable = False,
nullable=False,
)
scan_date = Column(
DateTime(timezone = True),
default = lambda: datetime.now(UTC),
nullable = False,
DateTime(timezone=True),
default=lambda: datetime.now(UTC),
nullable=False,
)
user = relationship("User", backref = "scans")
user = relationship("User", backref="scans")
test_results = relationship(
"TestResult",
back_populates = "scan",
cascade = "all, delete-orphan",
back_populates="scan",
cascade="all, delete-orphan",
)
def __repr__(self) -> str:
@ -64,10 +64,7 @@ class Scan(BaseModel):
Returns:
bool: True if any test result is vulnerable
"""
return any(
result.status == "vulnerable"
for result in self.test_results
)
return any(result.status == "vulnerable" for result in self.test_results)
@property
def vulnerability_count(self) -> int:
@ -77,7 +74,4 @@ class Scan(BaseModel):
Returns:
int: Number of vulnerable test results
"""
return sum(
1 for result in self.test_results
if result.status == "vulnerable"
)
return sum(1 for result in self.test_results if result.status == "vulnerable")

View File

@ -8,7 +8,7 @@ from sqlalchemy import (
Enum,
Integer,
Text,
ForeignKey,
ForeignKey,
)
from sqlalchemy.orm import relationship
from sqlalchemy.dialects.postgresql import JSON
@ -25,39 +25,35 @@ class TestResult(BaseModel):
"""
Stores individual test results for each security scan
"""
__tablename__ = "test_results"
scan_id = Column(
Integer,
ForeignKey("scans.id",
ondelete = "CASCADE"),
nullable = False,
index = True,
ForeignKey("scans.id", ondelete="CASCADE"),
nullable=False,
index=True,
)
test_name = Column(
Enum(TestType),
nullable = False,
index = True,
nullable=False,
index=True,
)
status = Column(
Enum(ScanStatus),
nullable = False,
index = True,
nullable=False,
index=True,
)
severity = Column(
Enum(Severity),
nullable = False,
index = True,
)
details = Column(Text, nullable = False)
evidence_json = Column(JSON, nullable = False, default = dict)
recommendations_json = Column(
JSON,
nullable = False,
default = list
nullable=False,
index=True,
)
details = Column(Text, nullable=False)
evidence_json = Column(JSON, nullable=False, default=dict)
recommendations_json = Column(JSON, nullable=False, default=list)
scan = relationship("Scan", back_populates = "test_results")
scan = relationship("Scan", back_populates="test_results")
def __repr__(self) -> str:
"""

View File

@ -17,16 +17,17 @@ class User(BaseModel):
"""
Stores authentication credentials and user information
"""
__tablename__ = "users"
email = Column(
String(settings.EMAIL_MAX_LENGTH),
unique = True,
nullable = False,
index = True,
unique=True,
nullable=False,
index=True,
)
hashed_password = Column(String, nullable = False)
is_active = Column(Boolean, default = True, nullable = False)
hashed_password = Column(String, nullable=False)
is_active = Column(Boolean, default=True, nullable=False)
def __repr__(self) -> str:
"""

View File

@ -56,19 +56,19 @@ build-backend = "setuptools.build_meta"
[tool.setuptools]
py-modules = [
"main",
"config",
"main",
"config",
"factory"
]
[tool.setuptools.packages.find]
include = [
"core",
"routes",
"models",
"schemas",
"services",
"scanners",
"routes",
"models",
"schemas",
"services",
"scanners",
"repositories"
]
@ -203,21 +203,21 @@ load-plugins = [
persistent = true
suggestion-mode = true
init-hook = "import sys; import os; sys.path.insert(0, os.getcwd())"
ignore = [
"venv",
".venv",
"__pycache__",
"build",
"dist",
ignore = [
"venv",
".venv",
"__pycache__",
"build",
"dist",
".git",
".pytest_cache",
".mypy_cache",
".ruff_cache",
".pytest_cache",
".mypy_cache",
".ruff_cache",
]
ignore-paths = [
"^venv/.*",
"^.venv/.*",
"^build/.*",
ignore-paths = [
"^venv/.*",
"^.venv/.*",
"^build/.*",
"^dist/.*",
]
[tool.pylint.type-check]
@ -268,7 +268,7 @@ max-statements = 40
[tool.bandit]
exclude_dirs = [
".venv",
".venv",
"venv",
]
skips = ["B104"]

View File

@ -19,13 +19,9 @@ class ScanRepository:
"""
Repository for Scan database operations
"""
@staticmethod
def create_scan(
db: Session,
user_id: int,
target_url: str,
commit: bool = True
) -> Scan:
def create_scan(db: Session, user_id: int, target_url: str, commit: bool = True) -> Scan:
"""
Create a new scan
@ -39,9 +35,9 @@ class ScanRepository:
Scan: Created scan instance
"""
scan = Scan(
user_id = user_id,
target_url = target_url,
scan_date = datetime.now(UTC),
user_id=user_id,
target_url=target_url,
scan_date=datetime.now(UTC),
)
db.add(scan)
if commit:
@ -62,17 +58,15 @@ class ScanRepository:
Scan | None: Scan instance or None if not found
"""
return (
db.query(Scan).options(
joinedload(Scan.test_results)
).filter(Scan.id == scan_id).first()
db.query(Scan)
.options(joinedload(Scan.test_results))
.filter(Scan.id == scan_id)
.first()
)
@staticmethod
def get_by_user(
db: Session,
user_id: int,
skip: int = 0,
limit: int | None = None
db: Session, user_id: int, skip: int = 0, limit: int | None = None
) -> list[Scan]:
"""
Get all scans for a user with pagination.
@ -90,16 +84,17 @@ class ScanRepository:
limit = settings.DEFAULT_PAGINATION_LIMIT
return (
db.query(Scan).options(
joinedload(Scan.test_results)
).filter(Scan.user_id == user_id).order_by(
Scan.scan_date.desc()
).offset(skip).limit(limit).all()
db.query(Scan)
.options(joinedload(Scan.test_results))
.filter(Scan.user_id == user_id)
.order_by(Scan.scan_date.desc())
.offset(skip)
.limit(limit)
.all()
)
@staticmethod
def get_recent(db: Session,
limit: int | None = None) -> list[Scan]:
def get_recent(db: Session, limit: int | None = None) -> list[Scan]:
"""
Get most recent scans across all users.
@ -114,17 +109,15 @@ class ScanRepository:
limit = settings.DEFAULT_PAGINATION_LIMIT
return (
db.query(Scan).options(
joinedload(Scan.test_results)
).order_by(Scan.scan_date.desc()).limit(limit).all()
db.query(Scan)
.options(joinedload(Scan.test_results))
.order_by(Scan.scan_date.desc())
.limit(limit)
.all()
)
@staticmethod
def delete(
db: Session,
scan_id: int,
commit: bool = True
) -> bool:
def delete(db: Session, scan_id: int, commit: bool = True) -> bool:
"""
Delete a scan (cascades to test results).

View File

@ -21,6 +21,7 @@ class TestResultRepository:
"""
Repository for TestResult database operations
"""
@staticmethod
def create_test_result(
db: Session,
@ -30,8 +31,7 @@ class TestResultRepository:
status: ScanStatus,
severity: Severity,
details: str,
evidence_json: dict[str,
Any],
evidence_json: dict[str, Any],
recommendations_json: list[str],
commit: bool = True,
) -> TestResult:
@ -53,13 +53,13 @@ class TestResultRepository:
TestResult: Created test result instance
"""
test_result = TestResult(
scan_id = scan_id,
test_name = test_name,
status = status,
severity = severity,
details = details,
evidence_json = evidence_json,
recommendations_json = recommendations_json,
scan_id=scan_id,
test_name=test_name,
status=status,
severity=severity,
details=details,
evidence_json=evidence_json,
recommendations_json=recommendations_json,
)
db.add(test_result)
if commit:
@ -69,9 +69,7 @@ class TestResultRepository:
@staticmethod
def bulk_create(
db: Session,
test_results: list[TestResult],
commit: bool = True
db: Session, test_results: list[TestResult], commit: bool = True
) -> list[TestResult]:
"""
Create multiple test results in bulk
@ -104,15 +102,14 @@ class TestResultRepository:
list[TestResult]: List of test results for the scan
"""
return (
db.query(TestResult).filter(
TestResult.scan_id == scan_id
).order_by(TestResult.created_at.asc()).all()
db.query(TestResult)
.filter(TestResult.scan_id == scan_id)
.order_by(TestResult.created_at.asc())
.all()
)
@staticmethod
def get_by_status(db: Session,
scan_id: int,
status: ScanStatus) -> list[TestResult]:
def get_by_status(db: Session, scan_id: int, status: ScanStatus) -> list[TestResult]:
"""
Get test results by status for a scan
@ -125,15 +122,13 @@ class TestResultRepository:
list[TestResult]: Filtered test results
"""
return (
db.query(TestResult).filter(
TestResult.scan_id == scan_id,
TestResult.status == status
).all()
db.query(TestResult)
.filter(TestResult.scan_id == scan_id, TestResult.status == status)
.all()
)
@staticmethod
def get_vulnerabilities(db: Session,
scan_id: int) -> list[TestResult]:
def get_vulnerabilities(db: Session, scan_id: int) -> list[TestResult]:
"""
Get only vulnerable test results for a scan
@ -144,18 +139,10 @@ class TestResultRepository:
Returns:
list[TestResult]: Vulnerable test results only
"""
return TestResultRepository.get_by_status(
db,
scan_id,
ScanStatus.VULNERABLE
)
return TestResultRepository.get_by_status(db, scan_id, ScanStatus.VULNERABLE)
@staticmethod
def delete_by_scan(
db: Session,
scan_id: int,
commit: bool = True
) -> int:
def delete_by_scan(db: Session, scan_id: int, commit: bool = True) -> int:
"""
Delete all test results for a scan
@ -167,11 +154,7 @@ class TestResultRepository:
Returns:
int: Number of test results deleted
"""
count = (
db.query(TestResult).filter(
TestResult.scan_id == scan_id
).delete()
)
count = db.query(TestResult).filter(TestResult.scan_id == scan_id).delete()
if commit:
db.commit()
return count

View File

@ -15,6 +15,7 @@ class UserRepository:
"""
Repository for User database operations
"""
@staticmethod
def get_by_id(db: Session, user_id: int) -> User | None:
"""
@ -45,10 +46,7 @@ class UserRepository:
@staticmethod
def create_user(
db: Session,
email: str,
hashed_password: str,
commit: bool = True
db: Session, email: str, hashed_password: str, commit: bool = True
) -> User:
"""
Create a new user
@ -62,7 +60,7 @@ class UserRepository:
Returns:
User: Created user instance
"""
user = User(email = email, hashed_password = hashed_password)
user = User(email=email, hashed_password=hashed_password)
db.add(user)
if commit:
db.commit()
@ -70,11 +68,7 @@ class UserRepository:
return user
@staticmethod
def get_all_active(
db: Session,
skip: int = 0,
limit: int | None = None
) -> list[User]:
def get_all_active(db: Session, skip: int = 0, limit: int | None = None) -> list[User]:
"""
Get all active users with pagination
@ -89,17 +83,11 @@ class UserRepository:
if limit is None:
limit = settings.DEFAULT_PAGINATION_LIMIT
return (
db.query(User).filter(User.is_active
).offset(skip).limit(limit).all()
)
return db.query(User).filter(User.is_active).offset(skip).limit(limit).all()
@staticmethod
def update_active_status(
db: Session,
user_id: int,
is_active: bool,
commit: bool = True
db: Session, user_id: int, is_active: bool, commit: bool = True
) -> User | None:
"""
Update user active status
@ -122,11 +110,7 @@ class UserRepository:
return user
@staticmethod
def delete(
db: Session,
user_id: int,
commit: bool = True
) -> bool:
def delete(db: Session, user_id: int, commit: bool = True) -> bool:
"""
Delete a user

View File

@ -24,14 +24,14 @@ from schemas.user_schemas import (
from services.auth_service import AuthService
router = APIRouter(prefix = "/auth", tags = ["authentication"])
limiter = Limiter(key_func = get_remote_address)
router = APIRouter(prefix="/auth", tags=["authentication"])
limiter = Limiter(key_func=get_remote_address)
@router.post(
"/register",
response_model = UserResponse,
status_code = status.HTTP_201_CREATED,
response_model=UserResponse,
status_code=status.HTTP_201_CREATED,
)
@limiter.limit(settings.API_RATE_LIMIT_REGISTER)
async def register(
@ -47,8 +47,8 @@ async def register(
@router.post(
"/login",
response_model = TokenResponse,
status_code = status.HTTP_200_OK,
response_model=TokenResponse,
status_code=status.HTTP_200_OK,
)
@limiter.limit(settings.API_RATE_LIMIT_LOGIN)
async def login(

View File

@ -24,14 +24,14 @@ from schemas.user_schemas import UserResponse
from services.scan_service import ScanService
router = APIRouter(prefix = "/scans", tags = ["scans"])
limiter = Limiter(key_func = get_remote_address)
router = APIRouter(prefix="/scans", tags=["scans"])
limiter = Limiter(key_func=get_remote_address)
@router.post(
"/",
response_model = ScanResponse,
status_code = status.HTTP_201_CREATED,
response_model=ScanResponse,
status_code=status.HTTP_201_CREATED,
)
@limiter.limit(settings.API_RATE_LIMIT_SCAN)
async def create_scan(
@ -48,8 +48,8 @@ async def create_scan(
@router.get(
"/",
response_model = list[ScanResponse],
status_code = status.HTTP_200_OK,
response_model=list[ScanResponse],
status_code=status.HTTP_200_OK,
)
@limiter.limit(settings.API_RATE_LIMIT_DEFAULT)
async def get_user_scans(
@ -62,18 +62,13 @@ async def get_user_scans(
"""
Get all scans for the authenticated user
"""
return ScanService.get_user_scans(
db,
current_user.id,
skip,
limit
)
return ScanService.get_user_scans(db, current_user.id, skip, limit)
@router.get(
"/{scan_id}",
response_model = ScanResponse,
status_code = status.HTTP_200_OK,
response_model=ScanResponse,
status_code=status.HTTP_200_OK,
)
@limiter.limit(settings.API_RATE_LIMIT_DEFAULT)
async def get_scan(
@ -90,7 +85,7 @@ async def get_scan(
@router.delete(
"/{scan_id}",
status_code = status.HTTP_204_NO_CONTENT,
status_code=status.HTTP_204_NO_CONTENT,
)
@limiter.limit(settings.API_RATE_LIMIT_DEFAULT)
async def delete_scan(

View File

@ -35,6 +35,7 @@ class AuthScanner(BaseScanner):
Maps to OWASP API Security Top 10 2023: API2:2023
"""
def scan(self) -> TestResultCreate:
"""
Execute authentication tests
@ -45,10 +46,10 @@ class AuthScanner(BaseScanner):
missing_auth_test = self._test_missing_authentication()
if missing_auth_test["vulnerable"]:
return self._create_vulnerable_result(
details = "Endpoint accessible without authentication",
evidence = missing_auth_test,
severity = Severity.HIGH,
recommendations = [
details="Endpoint accessible without authentication",
evidence=missing_auth_test,
severity=Severity.HIGH,
recommendations=[
"Require authentication for all sensitive endpoints",
"Implement proper authentication middleware",
"Return 401 Unauthorized for missing/invalid credentials",
@ -59,11 +60,10 @@ class AuthScanner(BaseScanner):
jwt_test = self._test_jwt_vulnerabilities()
if jwt_test["vulnerable"]:
return self._create_vulnerable_result(
details =
f"JWT vulnerability: {jwt_test['vulnerability_type']}",
evidence = jwt_test,
severity = Severity.CRITICAL,
recommendations = jwt_test.get(
details=f"JWT vulnerability: {jwt_test['vulnerability_type']}",
evidence=jwt_test,
severity=Severity.CRITICAL,
recommendations=jwt_test.get(
"recommendations",
[
"Properly validate JWT signatures",
@ -77,10 +77,10 @@ class AuthScanner(BaseScanner):
invalid_token_test = self._test_invalid_token_handling()
if invalid_token_test["vulnerable"]:
return self._create_vulnerable_result(
details = "Invalid tokens accepted by endpoint",
evidence = invalid_token_test,
severity = Severity.HIGH,
recommendations = [
details="Invalid tokens accepted by endpoint",
evidence=invalid_token_test,
severity=Severity.HIGH,
recommendations=[
"Reject invalid/malformed tokens with 401 status",
"Validate token format, signature, and expiration",
"Log authentication failures for monitoring",
@ -88,15 +88,15 @@ class AuthScanner(BaseScanner):
)
return TestResultCreate(
test_name = TestType.AUTH,
status = ScanStatus.SAFE,
severity = Severity.INFO,
details = "Authentication properly implemented",
evidence_json = {
test_name=TestType.AUTH,
status=ScanStatus.SAFE,
severity=Severity.INFO,
details="Authentication properly implemented",
evidence_json={
"missing_auth_test": missing_auth_test,
"invalid_token_test": invalid_token_test,
},
recommendations_json = [
recommendations_json=[
"Authentication is properly configured",
"Consider implementing additional security measures (2FA, refresh tokens)",
],
@ -114,8 +114,7 @@ class AuthScanner(BaseScanner):
session_without_auth = self.session.__class__()
session_without_auth.headers.update(
{
"User-Agent":
f"{settings.APP_NAME}/{settings.VERSION}",
"User-Agent": f"{settings.APP_NAME}/{settings.VERSION}",
"Accept": "application/json",
}
)
@ -123,29 +122,22 @@ class AuthScanner(BaseScanner):
try:
response = session_without_auth.get(
self.target_url,
timeout = settings.SCANNER_CONNECTION_TIMEOUT,
timeout=settings.SCANNER_CONNECTION_TIMEOUT,
)
if response.status_code == 200:
return {
"vulnerable":
True,
"status_code":
response.status_code,
"response_length":
len(response.text),
"description":
"Endpoint accessible without authentication",
"vulnerable": True,
"status_code": response.status_code,
"response_length": len(response.text),
"description": "Endpoint accessible without authentication",
}
if response.status_code in (401, 403):
return {
"vulnerable":
False,
"status_code":
response.status_code,
"description":
"Endpoint properly requires authentication",
"vulnerable": False,
"status_code": response.status_code,
"description": "Endpoint properly requires authentication",
}
return {
@ -158,8 +150,7 @@ class AuthScanner(BaseScanner):
return {
"vulnerable": False,
"error": str(e),
"description":
"Error testing authentication requirement",
"description": "Error testing authentication requirement",
}
def _test_jwt_vulnerabilities(self) -> dict[str, Any]:
@ -190,9 +181,7 @@ class AuthScanner(BaseScanner):
return {
"vulnerable": False,
"tests_performed":
["none_algorithm",
"signature_removal"],
"tests_performed": ["none_algorithm", "signature_removal"],
"description": "No JWT vulnerabilities detected",
}
@ -212,10 +201,7 @@ class AuthScanner(BaseScanner):
for variant in none_variants:
malicious_header = self._base64url_encode(
json.dumps({
"alg": variant,
"typ": "JWT"
})
json.dumps({"alg": variant, "typ": "JWT"})
)
malicious_token = f"{malicious_header}.{payload}."
@ -223,21 +209,15 @@ class AuthScanner(BaseScanner):
response = self.make_request(
"GET",
"/",
headers = {
"Authorization": f"Bearer {malicious_token}"
},
headers={"Authorization": f"Bearer {malicious_token}"},
)
if response.status_code == 200:
return {
"vulnerable":
True,
"vulnerability_type":
"JWT None Algorithm",
"algorithm_variant":
variant,
"status_code":
response.status_code,
"vulnerable": True,
"vulnerability_type": "JWT None Algorithm",
"algorithm_variant": variant,
"status_code": response.status_code,
"recommendations": [
"Reject tokens with 'none' algorithm (all case variations)",
"Explicitly verify signature before accepting tokens",
@ -272,19 +252,14 @@ class AuthScanner(BaseScanner):
response = self.make_request(
"GET",
"/",
headers = {
"Authorization": f"Bearer {malicious_token}"
},
headers={"Authorization": f"Bearer {malicious_token}"},
)
if response.status_code == 200:
return {
"vulnerable":
True,
"vulnerability_type":
"JWT Signature Not Verified",
"status_code":
response.status_code,
"vulnerable": True,
"vulnerability_type": "JWT Signature Not Verified",
"status_code": response.status_code,
"recommendations": [
"Require valid signature on all JWT tokens",
"Reject tokens with missing or invalid signatures",
@ -320,15 +295,13 @@ class AuthScanner(BaseScanner):
response = self.make_request(
"GET",
"/",
headers = {
"Authorization": f"Bearer {invalid_token}"
},
headers={"Authorization": f"Bearer {invalid_token}"},
)
if response.status_code == 200:
accepted_invalid.append(
{
"token": invalid_token[: 50],
"token": invalid_token[:50],
"status_code": response.status_code,
}
)
@ -382,8 +355,7 @@ class AuthScanner(BaseScanner):
def _create_vulnerable_result(
self,
details: str,
evidence: dict[str,
Any],
evidence: dict[str, Any],
severity: Severity = Severity.HIGH,
recommendations: list[str] | None = None,
) -> TestResultCreate:
@ -400,10 +372,10 @@ class AuthScanner(BaseScanner):
TestResultCreate: Vulnerable result
"""
return TestResultCreate(
test_name = TestType.AUTH,
status = ScanStatus.VULNERABLE,
severity = severity,
details = details,
evidence_json = evidence,
recommendations_json = recommendations or [],
test_name=TestType.AUTH,
status=ScanStatus.VULNERABLE,
severity=severity,
details=details,
evidence_json=evidence,
recommendations_json=recommendations or [],
)

View File

@ -25,6 +25,7 @@ class BaseScanner(ABC):
Provides common HTTP functionality, request spacing, retry logic,
and evidence collection. Specific scanners inherit and implement scan().
"""
def __init__(
self,
target_url: str,
@ -57,23 +58,17 @@ class BaseScanner(ABC):
session.headers.update(
{
"User-Agent":
f"{settings.APP_NAME}/{settings.VERSION}",
"User-Agent": f"{settings.APP_NAME}/{settings.VERSION}",
"Accept": "application/json",
}
)
if self.auth_token:
session.headers.update(
{"Authorization": f"Bearer {self.auth_token}"}
)
session.headers.update({"Authorization": f"Bearer {self.auth_token}"})
return session
def _wait_before_request(
self,
jitter_ms: int | None = None
) -> None:
def _wait_before_request(self, jitter_ms: int | None = None) -> None:
"""
Implement request spacing to avoid overwhelming target
@ -86,10 +81,7 @@ class BaseScanner(ABC):
if jitter_ms is None:
jitter_ms = settings.DEFAULT_JITTER_MS
required_delay = 1.0 / (
self.max_requests /
settings.SCANNER_RATE_LIMIT_WINDOW_SECONDS
)
required_delay = 1.0 / (self.max_requests / settings.SCANNER_RATE_LIMIT_WINDOW_SECONDS)
jitter = random.uniform(0, jitter_ms / 1000.0)
elapsed = time.time() - self.last_request_time
@ -130,31 +122,24 @@ class BaseScanner(ABC):
retry_count = 0
backoff_factor = 2.0
kwargs.setdefault(
"timeout",
settings.SCANNER_CONNECTION_TIMEOUT
)
kwargs.setdefault("timeout", settings.SCANNER_CONNECTION_TIMEOUT)
while retry_count <= settings.DEFAULT_RETRY_COUNT:
try:
start_time = time.time()
response = self.session.request(method, url, **kwargs)
setattr(
response,
"request_time",
time.time() - start_time
)
setattr(response, "request_time", time.time() - start_time)
self.request_count += 1
if response.status_code == 429:
retry_after = response.headers.get(
"Retry-After",
str(settings.DEFAULT_RETRY_WAIT_SECONDS)
"Retry-After", str(settings.DEFAULT_RETRY_WAIT_SECONDS)
)
wait_time = (
int(retry_after) if retry_after.isdigit() else
settings.DEFAULT_RETRY_WAIT_SECONDS
int(retry_after)
if retry_after.isdigit()
else settings.DEFAULT_RETRY_WAIT_SECONDS
)
time.sleep(wait_time)
retry_count += 1
@ -179,11 +164,8 @@ class BaseScanner(ABC):
return response
def get_baseline_timing(
self,
endpoint: str,
samples: int | None = None
) -> tuple[float,
float]:
self, endpoint: str, samples: int | None = None
) -> tuple[float, float]:
"""
Establish baseline response time for an endpoint
@ -214,8 +196,7 @@ class BaseScanner(ABC):
response: requests.Response,
payload: Any | None = None,
**additional_data: Any,
) -> dict[str,
Any]:
) -> dict[str, Any]:
"""
Collect evidence from test execution with sensitive data redaction
@ -228,17 +209,10 @@ class BaseScanner(ABC):
dict[str, Any]: Evidence dictionary
"""
evidence = {
"status_code":
response.status_code,
"response_time_ms":
round(getattr(response,
"request_time",
0.0) * 1000,
2),
"response_length":
len(response.text),
"headers":
self._redact_sensitive_headers(dict(response.headers)),
"status_code": response.status_code,
"response_time_ms": round(getattr(response, "request_time", 0.0) * 1000, 2),
"response_length": len(response.text),
"headers": self._redact_sensitive_headers(dict(response.headers)),
}
if payload is not None:
@ -248,10 +222,7 @@ class BaseScanner(ABC):
return evidence
def _redact_sensitive_headers(self,
headers: dict[str,
str]) -> dict[str,
str]:
def _redact_sensitive_headers(self, headers: dict[str, str]) -> dict[str, str]:
"""
Redact sensitive header values for evidence collection

View File

@ -30,6 +30,7 @@ class IDORScanner(BaseScanner):
Maps to OWASP API Security Top 10 2023: API1:2023
"""
def scan(self) -> TestResultCreate:
"""
Execute IDOR/BOLA tests
@ -41,11 +42,10 @@ class IDORScanner(BaseScanner):
if id_enumeration_test["vulnerable"]:
return self._create_vulnerable_result(
details =
f"IDOR vulnerability detected: {id_enumeration_test['vulnerability_type']}",
evidence = id_enumeration_test,
severity = Severity.HIGH,
recommendations = [
details=f"IDOR vulnerability detected: {id_enumeration_test['vulnerability_type']}",
evidence=id_enumeration_test,
severity=Severity.HIGH,
recommendations=[
"Implement proper authorization checks for all object access",
"Verify user owns/has permission to access requested resource",
"Use UUIDs instead of sequential IDs (but still check authorization)",
@ -58,11 +58,10 @@ class IDORScanner(BaseScanner):
if predictable_id_test["vulnerable"]:
return self._create_vulnerable_result(
details =
"Predictable ID patterns detected enabling enumeration",
evidence = predictable_id_test,
severity = Severity.MEDIUM,
recommendations = [
details="Predictable ID patterns detected enabling enumeration",
evidence=predictable_id_test,
severity=Severity.MEDIUM,
recommendations=[
"Use non-sequential, non-predictable identifiers (UUIDs)",
"Implement rate limiting on ID-based endpoints",
"Add authorization checks regardless of ID format",
@ -70,15 +69,15 @@ class IDORScanner(BaseScanner):
)
return TestResultCreate(
test_name = TestType.IDOR,
status = ScanStatus.SAFE,
severity = Severity.INFO,
details = "No IDOR/BOLA vulnerabilities detected",
evidence_json = {
test_name=TestType.IDOR,
status=ScanStatus.SAFE,
severity=Severity.INFO,
details="No IDOR/BOLA vulnerabilities detected",
evidence_json={
"id_enumeration_test": id_enumeration_test,
"predictable_id_test": predictable_id_test,
},
recommendations_json = [
recommendations_json=[
"Authorization checks appear to be in place",
"Continue monitoring for authorization bypasses",
],
@ -102,9 +101,7 @@ class IDORScanner(BaseScanner):
"description": "No IDs found in endpoint responses",
}
numeric_test = self._test_numeric_id_manipulation(
extracted_ids
)
numeric_test = self._test_numeric_id_manipulation(extracted_ids)
if numeric_test["vulnerable"]:
return numeric_test
@ -135,32 +132,22 @@ class IDORScanner(BaseScanner):
response_text = response.text
uuid_pattern = r'[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}'
uuids = re.findall(
uuid_pattern,
response_text,
re.IGNORECASE
)
uuid_pattern = r"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}"
uuids = re.findall(uuid_pattern, response_text, re.IGNORECASE)
numeric_id_pattern = r'"id"\s*:\s*(\d+)'
numeric_ids = re.findall(
numeric_id_pattern,
response_text
)
numeric_ids = re.findall(numeric_id_pattern, response_text)
ids = []
ids.extend(uuids[: 3])
ids.extend([int(nid) for nid in numeric_ids[: 3]])
ids.extend(uuids[:3])
ids.extend([int(nid) for nid in numeric_ids[:3]])
return ids
except Exception:
return []
def _test_numeric_id_manipulation(self,
extracted_ids: list[Any]
) -> dict[str,
Any]:
def _test_numeric_id_manipulation(self, extracted_ids: list[Any]) -> dict[str, Any]:
"""
Test numeric ID manipulation for IDOR
@ -170,10 +157,7 @@ class IDORScanner(BaseScanner):
Returns:
dict[str, Any]: Numeric ID manipulation test results
"""
numeric_ids = [
id_val for id_val in extracted_ids
if isinstance(id_val, int)
]
numeric_ids = [id_val for id_val in extracted_ids if isinstance(id_val, int)]
if not numeric_ids:
return {
@ -220,10 +204,7 @@ class IDORScanner(BaseScanner):
"numeric_ids_tested": len(test_ids),
}
def _test_string_id_manipulation(self,
extracted_ids: list[Any]
) -> dict[str,
Any]:
def _test_string_id_manipulation(self, extracted_ids: list[Any]) -> dict[str, Any]:
"""
Test string/UUID ID manipulation for IDOR
@ -233,10 +214,7 @@ class IDORScanner(BaseScanner):
Returns:
dict[str, Any]: String ID manipulation test results
"""
string_ids = [
id_val for id_val in extracted_ids
if isinstance(id_val, str)
]
string_ids = [id_val for id_val in extracted_ids if isinstance(id_val, str)]
if not string_ids:
return {
@ -305,7 +283,7 @@ class IDORScanner(BaseScanner):
"vulnerable": True,
"pattern_type": "Sequential IDs",
"id_difference": diff1,
"example_ids": numeric_ids1[: 3],
"example_ids": numeric_ids1[:3],
}
return {
@ -323,8 +301,7 @@ class IDORScanner(BaseScanner):
def _create_vulnerable_result(
self,
details: str,
evidence: dict[str,
Any],
evidence: dict[str, Any],
severity: Severity = Severity.HIGH,
recommendations: list[str] | None = None,
) -> TestResultCreate:
@ -341,10 +318,10 @@ class IDORScanner(BaseScanner):
TestResultCreate: Vulnerable result
"""
return TestResultCreate(
test_name = TestType.IDOR,
status = ScanStatus.VULNERABLE,
severity = severity,
details = details,
evidence_json = evidence,
recommendations_json = recommendations or [],
test_name=TestType.IDOR,
status=ScanStatus.VULNERABLE,
severity=severity,
details=details,
evidence_json=evidence,
recommendations_json=recommendations or [],
)

View File

@ -8,6 +8,7 @@ class SQLiPayloads:
"""
SQL Injection test payloads covering various database types and techniques
"""
ERROR_SIGNATURES = {
"mysql": [
"sql syntax",
@ -130,10 +131,13 @@ class SQLiPayloads:
list[str]: All SQLi test payloads
"""
return (
cls.BASIC_AUTHENTICATION_BYPASS + cls.UNION_BASED +
cls.TIME_BASED_BLIND + cls.BOOLEAN_BASED_BLIND +
cls.ERROR_BASED + cls.STACKED_QUERIES +
cls.COMMENT_VARIATIONS
cls.BASIC_AUTHENTICATION_BYPASS
+ cls.UNION_BASED
+ cls.TIME_BASED_BLIND
+ cls.BOOLEAN_BASED_BLIND
+ cls.ERROR_BASED
+ cls.STACKED_QUERIES
+ cls.COMMENT_VARIATIONS
)
@classmethod
@ -221,6 +225,7 @@ class IDORPayloads:
"""
Insecure Direct Object Reference (IDOR) test patterns
"""
NUMERIC_ID_MANIPULATIONS = [
0,
-1,
@ -273,13 +278,11 @@ class RateLimitBypassPayloads:
"""
Rate limiting bypass techniques and patterns
"""
HEADER_PATTERNS = {
"limit":
r"x-ratelimit-limit|x-rate-limit-limit|ratelimit-limit",
"remaining":
r"x-ratelimit-remaining|x-rate-limit-remaining|ratelimit-remaining",
"reset":
r"x-ratelimit-reset|x-rate-limit-reset|ratelimit-reset",
"limit": r"x-ratelimit-limit|x-rate-limit-limit|ratelimit-limit",
"remaining": r"x-ratelimit-remaining|x-rate-limit-remaining|ratelimit-remaining",
"reset": r"x-ratelimit-reset|x-rate-limit-reset|ratelimit-reset",
"retry_after": r"retry-after",
}
@ -297,30 +300,14 @@ class RateLimitBypassPayloads:
]
HEADER_SPOOFING = [
{
"X-Forwarded-For": "127.0.0.1"
},
{
"X-Forwarded-For": "8.8.8.8"
},
{
"X-Real-IP": "127.0.0.1"
},
{
"X-Originating-IP": "127.0.0.1"
},
{
"X-Remote-IP": "127.0.0.1"
},
{
"X-Client-IP": "127.0.0.1"
},
{
"CF-Connecting-IP": "127.0.0.1"
},
{
"True-Client-IP": "127.0.0.1"
},
{"X-Forwarded-For": "127.0.0.1"},
{"X-Forwarded-For": "8.8.8.8"},
{"X-Real-IP": "127.0.0.1"},
{"X-Originating-IP": "127.0.0.1"},
{"X-Remote-IP": "127.0.0.1"},
{"X-Client-IP": "127.0.0.1"},
{"CF-Connecting-IP": "127.0.0.1"},
{"True-Client-IP": "127.0.0.1"},
]
USER_AGENT_ROTATION = [
@ -413,9 +400,9 @@ class XSSPayloads:
ATTRIBUTE_BREAKING = [
"' onmouseover='alert(\"XSS\")'",
"\" onmouseover=\"alert('XSS')\"",
'" onmouseover="alert(\'XSS\')"',
"' onclick='alert(\"XSS\")' '",
"\" autofocus onfocus=\"alert('XSS')\"",
'" autofocus onfocus="alert(\'XSS\')"',
"'/><script>alert('XSS')</script>",
"\"/><script>alert('XSS')</script>",
]
@ -427,7 +414,7 @@ class XSSPayloads:
"<script>alert('XSS')<!--",
"<<script>alert('XSS')</script>",
"<script\x20type='text/javascript'>alert('XSS')</script>",
"<script\x0D\x0A>alert('XSS')</script>",
"<script\x0d\x0a>alert('XSS')</script>",
]
POLYGLOT_XSS = [
@ -446,10 +433,14 @@ class XSSPayloads:
list[str]: All XSS test payloads
"""
return (
cls.BASIC_XSS + cls.EVENT_HANDLER_XSS + cls.SVG_XSS +
cls.IFRAME_XSS + cls.ENCODED_XSS +
cls.ATTRIBUTE_BREAKING + cls.FILTER_BYPASS +
cls.POLYGLOT_XSS
cls.BASIC_XSS
+ cls.EVENT_HANDLER_XSS
+ cls.SVG_XSS
+ cls.IFRAME_XSS
+ cls.ENCODED_XSS
+ cls.ATTRIBUTE_BREAKING
+ cls.FILTER_BYPASS
+ cls.POLYGLOT_XSS
)
@classmethod

View File

@ -26,6 +26,7 @@ class RateLimitScanner(BaseScanner):
"""
Rate limiting and bypass vulnerabilities tests
"""
def scan(self) -> TestResultCreate:
"""
Execute rate limiting tests
@ -37,10 +38,9 @@ class RateLimitScanner(BaseScanner):
if not rate_limit_info["rate_limit_detected"]:
return self._create_vulnerable_result(
details =
"No rate limiting detected on target endpoint",
evidence = rate_limit_info,
recommendations = [
details="No rate limiting detected on target endpoint",
evidence=rate_limit_info,
recommendations=[
"Implement rate limiting to prevent abuse and DoS attacks",
"Use standard rate limit headers (X-RateLimit-Limit, X-RateLimit-Remaining)",
"Return 429 Too Many Requests when limits are exceeded",
@ -50,11 +50,10 @@ class RateLimitScanner(BaseScanner):
if rate_limit_info["enforcement_status"] == "HEADERS_ONLY":
return self._create_vulnerable_result(
details =
"Rate limit headers present but not enforced",
evidence = rate_limit_info,
severity = Severity.MEDIUM,
recommendations = [
details="Rate limit headers present but not enforced",
evidence=rate_limit_info,
severity=Severity.MEDIUM,
recommendations=[
"Enforce rate limits with 429 responses when thresholds are exceeded",
"Rate limit headers without enforcement provide false security",
],
@ -64,14 +63,13 @@ class RateLimitScanner(BaseScanner):
if bypass_results["bypass_successful"]:
return self._create_vulnerable_result(
details =
f"Rate limiting bypassed using: {bypass_results['bypass_method']}",
evidence = {
details=f"Rate limiting bypassed using: {bypass_results['bypass_method']}",
evidence={
"rate_limit_info": rate_limit_info,
"bypass_details": bypass_results,
},
severity = Severity.HIGH,
recommendations = [
severity=Severity.HIGH,
recommendations=[
f"Fix bypass vulnerability: {bypass_results['bypass_method']}",
"Do not trust client-provided IP headers (X-Forwarded-For, X-Real-IP)",
"Implement rate limiting at multiple layers (IP, user, API key)",
@ -80,25 +78,21 @@ class RateLimitScanner(BaseScanner):
)
return TestResultCreate(
test_name = TestType.RATE_LIMIT,
status = ScanStatus.SAFE,
severity = Severity.INFO,
details =
"Rate limiting properly implemented and enforced",
evidence_json = {
test_name=TestType.RATE_LIMIT,
status=ScanStatus.SAFE,
severity=Severity.INFO,
details="Rate limiting properly implemented and enforced",
evidence_json={
"rate_limit_info": rate_limit_info,
"bypass_attempts": bypass_results,
},
recommendations_json = [
recommendations_json=[
"Rate limiting is properly configured",
"Continue monitoring for new bypass techniques",
],
)
def _detect_rate_limiting(self,
test_request_count: int = 20
) -> dict[str,
Any]:
def _detect_rate_limiting(self, test_request_count: int = 20) -> dict[str, Any]:
"""
Detect rate limiting by analyzing headers and response patterns
@ -110,8 +104,7 @@ class RateLimitScanner(BaseScanner):
Returns:
dict[str, Any]: Rate limiting detection results
"""
rate_limit_patterns = RateLimitBypassPayloads.get_header_patterns(
)
rate_limit_patterns = RateLimitBypassPayloads.get_header_patterns()
results = {
"rate_limit_detected": False,
@ -127,35 +120,23 @@ class RateLimitScanner(BaseScanner):
try:
response = self.make_request("GET", "/")
headers_lower = {
k.lower(): v
for k, v in response.headers.items()
}
headers_lower = {k.lower(): v for k, v in response.headers.items()}
for header_type, pattern in rate_limit_patterns.items():
for header_name, header_value in headers_lower.items():
if re.search(pattern,
header_name,
re.IGNORECASE):
results["rate_limit_headers"][
header_type] = {
"header_name": header_name,
"value": header_value,
}
if re.search(pattern, header_name, re.IGNORECASE):
results["rate_limit_headers"][header_type] = {
"header_name": header_name,
"value": header_value,
}
results["rate_limit_detected"] = True
results["request_results"].append(
{
"attempt":
attempt,
"status_code":
response.status_code,
"response_time_ms":
round(
getattr(response,
"request_time",
0.0) * 1000,
2
"attempt": attempt,
"status_code": response.status_code,
"response_time_ms": round(
getattr(response, "request_time", 0.0) * 1000, 2
),
}
)
@ -173,22 +154,15 @@ class RateLimitScanner(BaseScanner):
time.sleep(0.1)
except Exception as e:
results["request_results"].append(
{
"attempt": attempt,
"error": str(e)
}
)
results["request_results"].append({"attempt": attempt, "error": str(e)})
break
if results["rate_limit_detected"]:
if "limit" in results["rate_limit_headers"]:
results["limit_threshold"] = results[
"rate_limit_headers"]["limit"]["value"]
results["limit_threshold"] = results["rate_limit_headers"]["limit"]["value"]
if "reset" in results["rate_limit_headers"]:
results["reset_window"] = results["rate_limit_headers"
]["reset"]["value"]
results["reset_window"] = results["rate_limit_headers"]["reset"]["value"]
if not results["enforcement_status"]:
results["enforcement_status"] = "HEADERS_ONLY"
@ -235,9 +209,7 @@ class RateLimitScanner(BaseScanner):
return results
def _test_ip_header_bypass(self,
test_count: int = 15) -> dict[str,
Any]:
def _test_ip_header_bypass(self, test_count: int = 15) -> dict[str, Any]:
"""
Test if rate limiting can be bypassed with IP spoofing headers
@ -261,11 +233,7 @@ class RateLimitScanner(BaseScanner):
test_headers = {header_name: fake_ip}
try:
response = self.make_request(
"GET",
"/",
headers = test_headers
)
response = self.make_request("GET", "/", headers=test_headers)
if response.status_code != 429:
success_count += 1
@ -285,8 +253,7 @@ class RateLimitScanner(BaseScanner):
return {
"bypass_successful": False,
"headers_tested":
[list(h.keys())[0] for h in bypass_headers],
"headers_tested": [list(h.keys())[0] for h in bypass_headers],
}
def _test_endpoint_variation_bypass(self) -> dict[str, Any]:
@ -328,8 +295,7 @@ class RateLimitScanner(BaseScanner):
def _create_vulnerable_result(
self,
details: str,
evidence: dict[str,
Any],
evidence: dict[str, Any],
severity: Severity = Severity.HIGH,
recommendations: list[str] | None = None,
) -> TestResultCreate:
@ -346,10 +312,10 @@ class RateLimitScanner(BaseScanner):
TestResultCreate: Vulnerable result
"""
return TestResultCreate(
test_name = TestType.RATE_LIMIT,
status = ScanStatus.VULNERABLE,
severity = severity,
details = details,
evidence_json = evidence,
recommendations_json = recommendations or [],
test_name=TestType.RATE_LIMIT,
status=ScanStatus.VULNERABLE,
severity=severity,
details=details,
evidence_json=evidence,
recommendations_json=recommendations or [],
)

View File

@ -31,6 +31,7 @@ class SQLiScanner(BaseScanner):
Uses payloads covering MySQL, PostgreSQL, MSSQL, Oracle
"""
def scan(self) -> TestResultCreate:
"""
Execute SQL injection tests
@ -41,11 +42,10 @@ class SQLiScanner(BaseScanner):
error_based_test = self._test_error_based_sqli()
if error_based_test["vulnerable"]:
return self._create_vulnerable_result(
details =
f"Error-based SQL injection detected: {error_based_test['database_type']}",
evidence = error_based_test,
severity = Severity.CRITICAL,
recommendations = [
details=f"Error-based SQL injection detected: {error_based_test['database_type']}",
evidence=error_based_test,
severity=Severity.CRITICAL,
recommendations=[
"Use parameterized queries (prepared statements)",
"Never concatenate user input into SQL queries",
"Implement input validation and sanitization",
@ -57,10 +57,10 @@ class SQLiScanner(BaseScanner):
boolean_based_test = self._test_boolean_based_sqli()
if boolean_based_test["vulnerable"]:
return self._create_vulnerable_result(
details = "Boolean-based blind SQL injection detected",
evidence = boolean_based_test,
severity = Severity.CRITICAL,
recommendations = [
details="Boolean-based blind SQL injection detected",
evidence=boolean_based_test,
severity=Severity.CRITICAL,
recommendations=[
"Use parameterized queries for all database operations",
"Implement proper input validation",
"Avoid exposing different responses for true/false conditions",
@ -70,11 +70,10 @@ class SQLiScanner(BaseScanner):
time_based_test = self._test_time_based_sqli()
if time_based_test["vulnerable"]:
return self._create_vulnerable_result(
details =
f"Time-based blind SQL injection detected: {time_based_test['database_type']}",
evidence = time_based_test,
severity = Severity.CRITICAL,
recommendations = [
details=f"Time-based blind SQL injection detected: {time_based_test['database_type']}",
evidence=time_based_test,
severity=Severity.CRITICAL,
recommendations=[
"Use parameterized queries exclusively",
"Implement strict input validation",
"Monitor for unusual response time patterns",
@ -82,16 +81,16 @@ class SQLiScanner(BaseScanner):
)
return TestResultCreate(
test_name = TestType.SQLI,
status = ScanStatus.SAFE,
severity = Severity.INFO,
details = "No SQL injection vulnerabilities detected",
evidence_json = {
test_name=TestType.SQLI,
status=ScanStatus.SAFE,
severity=Severity.INFO,
details="No SQL injection vulnerabilities detected",
evidence_json={
"error_based_test": error_based_test,
"boolean_based_test": boolean_based_test,
"time_based_test": time_based_test,
},
recommendations_json = [
recommendations_json=[
"Continue using parameterized queries",
"Regularly update security testing",
],
@ -125,8 +124,7 @@ class SQLiScanner(BaseScanner):
"payload": payload,
"status_code": response.status_code,
"error_signature": signature,
"response_excerpt":
response.text[: 500],
"response_excerpt": response.text[:500],
}
except Exception:
@ -161,12 +159,12 @@ class SQLiScanner(BaseScanner):
boolean_payloads = SQLiPayloads.BOOLEAN_BASED_BLIND
true_payloads = [
p for p in boolean_payloads
if "AND '1'='1" in p or "AND 1=1" in p
p for p in boolean_payloads if "AND '1'='1" in p or "AND 1=1" in p
]
false_payloads = [
p for p in boolean_payloads if "AND '1'='2" in p
or "AND 1=2" in p or "AND 1=0" in p
p
for p in boolean_payloads
if "AND '1'='2" in p or "AND 1=2" in p or "AND 1=0" in p
]
true_lengths = []
@ -186,18 +184,12 @@ class SQLiScanner(BaseScanner):
if length_diff > 100 and avg_true != avg_false:
return {
"vulnerable":
True,
"baseline_length":
baseline_length,
"true_condition_avg_length":
avg_true,
"false_condition_avg_length":
avg_false,
"length_difference":
length_diff,
"confidence":
"HIGH" if length_diff > 500 else "MEDIUM",
"vulnerable": True,
"baseline_length": baseline_length,
"true_condition_avg_length": avg_true,
"false_condition_avg_length": avg_false,
"length_difference": length_diff,
"confidence": "HIGH" if length_diff > 500 else "MEDIUM",
}
return {
@ -213,9 +205,7 @@ class SQLiScanner(BaseScanner):
"description": "Error testing boolean-based SQLi",
}
def _test_time_based_sqli(self,
delay_seconds: int = 5) -> dict[str,
Any]:
def _test_time_based_sqli(self, delay_seconds: int = 5) -> dict[str, Any]:
"""
Test for time based blind SQL injection
@ -237,14 +227,9 @@ class SQLiScanner(BaseScanner):
all_time_payloads = SQLiPayloads.TIME_BASED_BLIND
delay_payloads = {
"mysql":
[p for p in all_time_payloads if "SLEEP" in p],
"postgres": [
p for p in all_time_payloads if "pg_sleep" in p
],
"mssql": [
p for p in all_time_payloads if "WAITFOR" in p
],
"mysql": [p for p in all_time_payloads if "SLEEP" in p],
"postgres": [p for p in all_time_payloads if "pg_sleep" in p],
"mssql": [p for p in all_time_payloads if "WAITFOR" in p],
}
for db_type, payloads in delay_payloads.items():
@ -256,13 +241,9 @@ class SQLiScanner(BaseScanner):
response = self.make_request(
"GET",
f"/?id={payload}",
timeout = delay_seconds + 10,
)
elapsed = getattr(
response,
"request_time",
0.0
timeout=delay_seconds + 10,
)
elapsed = getattr(response, "request_time", 0.0)
delay_times.append(elapsed)
except Exception:
@ -273,28 +254,17 @@ class SQLiScanner(BaseScanner):
avg_delay = statistics.mean(delay_times)
if avg_delay >= expected_delay_time - 1:
confidence = (
"HIGH" if avg_delay >= expected_delay_time
else "MEDIUM"
)
confidence = "HIGH" if avg_delay >= expected_delay_time else "MEDIUM"
return {
"vulnerable":
True,
"database_type":
db_type,
"payload":
payload,
"baseline_time":
f"{baseline_mean:.3f}s",
"response_time":
f"{avg_delay:.3f}s",
"expected_delay":
f"{expected_delay_time:.3f}s",
"confidence":
confidence,
"individual_times":
[f"{t:.3f}s" for t in delay_times],
"vulnerable": True,
"database_type": db_type,
"payload": payload,
"baseline_time": f"{baseline_mean:.3f}s",
"response_time": f"{avg_delay:.3f}s",
"expected_delay": f"{expected_delay_time:.3f}s",
"confidence": confidence,
"individual_times": [f"{t:.3f}s" for t in delay_times],
}
return {
@ -314,8 +284,7 @@ class SQLiScanner(BaseScanner):
def _create_vulnerable_result(
self,
details: str,
evidence: dict[str,
Any],
evidence: dict[str, Any],
severity: Severity = Severity.CRITICAL,
recommendations: list[str] | None = None,
) -> TestResultCreate:
@ -332,10 +301,10 @@ class SQLiScanner(BaseScanner):
TestResultCreate: Vulnerable result
"""
return TestResultCreate(
test_name = TestType.SQLI,
status = ScanStatus.VULNERABLE,
severity = severity,
details = details,
evidence_json = evidence,
recommendations_json = recommendations or [],
test_name=TestType.SQLI,
status=ScanStatus.VULNERABLE,
severity=severity,
details=details,
evidence_json=evidence,
recommendations_json=recommendations or [],
)

View File

@ -22,13 +22,14 @@ class ScanRequest(BaseModel):
"""
Schema for creating a new security scan
"""
target_url: HttpUrl = Field(max_length = settings.URL_MAX_LENGTH)
target_url: HttpUrl = Field(max_length=settings.URL_MAX_LENGTH)
auth_token: str | None = None
tests_to_run: list[TestType] = Field(min_length = 1)
tests_to_run: list[TestType] = Field(min_length=1)
max_requests: int = Field(
default = settings.DEFAULT_MAX_REQUESTS,
ge = 1,
le = settings.SCANNER_MAX_CONCURRENT_REQUESTS,
default=settings.DEFAULT_MAX_REQUESTS,
ge=1,
le=settings.SCANNER_MAX_CONCURRENT_REQUESTS,
)
@ -36,7 +37,8 @@ class ScanResponse(BaseModel):
"""
Schema for scan data in API responses
"""
model_config = ConfigDict(from_attributes = True)
model_config = ConfigDict(from_attributes=True)
id: int
user_id: int
@ -57,6 +59,4 @@ class ScanResponse(BaseModel):
"""
Number of vulnerabilities found
"""
return sum(
1 for r in self.test_results if r.status == "vulnerable"
)
return sum(1 for r in self.test_results if r.status == "vulnerable")

View File

@ -22,19 +22,21 @@ class TestResultCreate(BaseModel):
"""
Schema for creating a new test result (used by scanners)
"""
test_name: TestType
status: ScanStatus
severity: Severity
details: str
evidence_json: dict[str, Any] = Field(default_factory = dict)
recommendations_json: list[str] = Field(default_factory = list)
evidence_json: dict[str, Any] = Field(default_factory=dict)
recommendations_json: list[str] = Field(default_factory=list)
class TestResultResponse(BaseModel):
"""
Schema for individual test result in API responses
"""
model_config = ConfigDict(from_attributes = True)
model_config = ConfigDict(from_attributes=True)
id: int
scan_id: int

View File

@ -8,13 +8,7 @@ from __future__ import annotations
import re
from datetime import datetime
from pydantic import (
BaseModel,
ConfigDict,
EmailStr,
Field,
field_validator
)
from pydantic import BaseModel, ConfigDict, EmailStr, Field, field_validator
from config import settings
@ -22,24 +16,25 @@ class UserCreate(BaseModel):
"""
Schema for user registration request.
"""
email: EmailStr
password: str = Field(
min_length = settings.PASSWORD_MIN_LENGTH,
max_length = settings.PASSWORD_MAX_LENGTH,
min_length=settings.PASSWORD_MIN_LENGTH,
max_length=settings.PASSWORD_MAX_LENGTH,
)
@field_validator('password')
@field_validator("password")
@classmethod
def validate_password_strength(cls, v: str) -> str:
"""
Validate password meets security requirements
"""
if not re.search(r'[A-Z]', v):
raise ValueError('Password must contain at least one uppercase letter')
if not re.search(r'[a-z]', v):
raise ValueError('Password must contain at least one lowercase letter')
if not re.search(r'[0-9]', v):
raise ValueError('Password must contain at least one number')
if not re.search(r"[A-Z]", v):
raise ValueError("Password must contain at least one uppercase letter")
if not re.search(r"[a-z]", v):
raise ValueError("Password must contain at least one lowercase letter")
if not re.search(r"[0-9]", v):
raise ValueError("Password must contain at least one number")
return v
@ -47,6 +42,7 @@ class UserLogin(BaseModel):
"""
Schema for user login request.
"""
email: EmailStr
password: str
@ -56,7 +52,8 @@ class UserResponse(BaseModel):
Schema for user data in API responses.
Excludes sensitive fields like hashed_password.
"""
model_config = ConfigDict(from_attributes = True)
model_config = ConfigDict(from_attributes=True)
id: int
email: str
@ -68,5 +65,6 @@ class TokenResponse(BaseModel):
"""
Schema for JWT token response.
"""
access_token: str
token_type: str = "bearer"

View File

@ -28,11 +28,9 @@ class AuthService:
"""
User registration, login, and token generation
"""
@staticmethod
def register_user(
db: Session,
user_data: UserCreate
) -> UserResponse:
def register_user(db: Session, user_data: UserCreate) -> UserResponse:
"""
Register a new user
@ -43,31 +41,25 @@ class AuthService:
Returns:
UserResponse: Created user data
"""
existing_user = UserRepository.get_by_email(
db,
user_data.email
)
existing_user = UserRepository.get_by_email(db, user_data.email)
if existing_user:
raise HTTPException(
status_code = status.HTTP_400_BAD_REQUEST,
detail = "Email already registered",
status_code=status.HTTP_400_BAD_REQUEST,
detail="Email already registered",
)
hashed_password = hash_password(user_data.password)
user = UserRepository.create_user(
db = db,
email = user_data.email,
hashed_password = hashed_password,
db=db,
email=user_data.email,
hashed_password=hashed_password,
)
return UserResponse.model_validate(user)
@staticmethod
def login_user(
db: Session,
login_data: UserLogin
) -> TokenResponse:
def login_user(db: Session, login_data: UserLogin) -> TokenResponse:
"""
Authenticate user and generate access token
@ -82,40 +74,31 @@ class AuthService:
if not user:
raise HTTPException(
status_code = status.HTTP_401_UNAUTHORIZED,
detail = "Invalid email or password",
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid email or password",
)
if not verify_password(login_data.password,
user.hashed_password):
if not verify_password(login_data.password, user.hashed_password):
raise HTTPException(
status_code = status.HTTP_401_UNAUTHORIZED,
detail = "Invalid email or password",
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid email or password",
)
if not user.is_active:
raise HTTPException(
status_code = status.HTTP_403_FORBIDDEN,
detail = "Account is inactive",
status_code=status.HTTP_403_FORBIDDEN,
detail="Account is inactive",
)
access_token = create_access_token(
data = {"sub": user.email},
expires_delta = timedelta(
minutes = settings.ACCESS_TOKEN_EXPIRE_MINUTES
),
data={"sub": user.email},
expires_delta=timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES),
)
return TokenResponse(
access_token = access_token,
token_type = "bearer"
)
return TokenResponse(access_token=access_token, token_type="bearer")
@staticmethod
def get_user_by_email(
db: Session,
email: str
) -> UserResponse | None:
def get_user_by_email(db: Session, email: str) -> UserResponse | None:
"""
Get user by email address

View File

@ -5,8 +5,6 @@ Coordinates scanners and saves results
from __future__ import annotations
from typing import Type
from sqlalchemy.orm import Session
from fastapi import HTTPException, status
@ -27,12 +25,9 @@ class ScanService:
"""
Orchestrates security scanning workflow
"""
@staticmethod
def run_scan(
db: Session,
user_id: int,
scan_request: ScanRequest
) -> ScanResponse:
def run_scan(db: Session, user_id: int, scan_request: ScanRequest) -> ScanResponse:
"""
Execute security scan with selected tests
@ -45,12 +40,12 @@ class ScanService:
ScanResponse: Scan results with all test outcomes
"""
scan = ScanRepository.create_scan(
db = db,
user_id = user_id,
target_url = str(scan_request.target_url),
db=db,
user_id=user_id,
target_url=str(scan_request.target_url),
)
scanner_mapping: dict[TestType, Type[BaseScanner]] = {
scanner_mapping: dict[TestType, type[BaseScanner]] = {
TestType.RATE_LIMIT: RateLimitScanner,
TestType.AUTH: AuthScanner,
TestType.SQLI: SQLiScanner,
@ -60,16 +55,16 @@ class ScanService:
results: list[TestResultCreate] = []
for test_type in scan_request.tests_to_run:
scanner_class: Type[BaseScanner] | None = scanner_mapping.get(test_type)
scanner_class: type[BaseScanner] | None = scanner_mapping.get(test_type)
if not scanner_class:
continue
try:
scanner = scanner_class(
target_url = str(scan_request.target_url),
auth_token = scan_request.auth_token,
max_requests = scan_request.max_requests,
target_url=str(scan_request.target_url),
auth_token=scan_request.auth_token,
max_requests=scan_request.max_requests,
)
result = scanner.scan()
@ -78,12 +73,12 @@ class ScanService:
except Exception as e:
results.append(
TestResultCreate(
test_name = test_type,
status = "error",
severity = "info",
details = f"Scanner error: {str(e)}",
evidence_json = {"error": str(e)},
recommendations_json = [
test_name=test_type,
status="error",
severity="info",
details=f"Scanner error: {str(e)}",
evidence_json={"error": str(e)},
recommendations_json=[
"Check target URL is accessible",
"Verify authentication token if provided",
],
@ -92,14 +87,14 @@ class ScanService:
for result in results:
TestResultRepository.create_test_result(
db = db,
scan_id = scan.id,
test_name = result.test_name,
status = result.status,
severity = result.severity,
details = result.details,
evidence_json = result.evidence_json,
recommendations_json = result.recommendations_json,
db=db,
scan_id=scan.id,
test_name=result.test_name,
status=result.status,
severity=result.severity,
details=result.details,
evidence_json=result.evidence_json,
recommendations_json=result.recommendations_json,
)
db.refresh(scan)
@ -107,11 +102,7 @@ class ScanService:
return ScanResponse.model_validate(scan)
@staticmethod
def get_scan_by_id(
db: Session,
scan_id: int,
user_id: int
) -> ScanResponse:
def get_scan_by_id(db: Session, scan_id: int, user_id: int) -> ScanResponse:
"""
Get scan by ID with authorization check
@ -127,24 +118,21 @@ class ScanService:
if not scan:
raise HTTPException(
status_code = status.HTTP_404_NOT_FOUND,
detail = "Scan not found",
status_code=status.HTTP_404_NOT_FOUND,
detail="Scan not found",
)
if scan.user_id != user_id:
raise HTTPException(
status_code = status.HTTP_403_FORBIDDEN,
detail = "Not authorized to access this scan",
status_code=status.HTTP_403_FORBIDDEN,
detail="Not authorized to access this scan",
)
return ScanResponse.model_validate(scan)
@staticmethod
def get_user_scans(
db: Session,
user_id: int,
skip: int = 0,
limit: int | None = None
db: Session, user_id: int, skip: int = 0, limit: int | None = None
) -> list[ScanResponse]:
"""
Get all scans for a user with pagination
@ -158,12 +146,7 @@ class ScanService:
Returns:
list[ScanResponse]: List of user's scans
"""
scans = ScanRepository.get_by_user(
db = db,
user_id = user_id,
skip = skip,
limit = limit
)
scans = ScanRepository.get_by_user(db=db, user_id=user_id, skip=skip, limit=limit)
return [ScanResponse.model_validate(scan) for scan in scans]
@ -184,14 +167,14 @@ class ScanService:
if not scan:
raise HTTPException(
status_code = status.HTTP_404_NOT_FOUND,
detail = "Scan not found",
status_code=status.HTTP_404_NOT_FOUND,
detail="Scan not found",
)
if scan.user_id != user_id:
raise HTTPException(
status_code = status.HTTP_403_FORBIDDEN,
detail = "Not authorized to delete this scan",
status_code=status.HTTP_403_FORBIDDEN,
detail="Not authorized to delete this scan",
)
return ScanRepository.delete(db, scan_id)

View File

@ -20,7 +20,7 @@
{
"files": "*.tsx",
"options": {
"printWidth": 77,
"printWidth": 77,
"jsxSingleQuote": false,
"bracketSameLine": false
}
@ -28,7 +28,7 @@
{
"files": "*.ts",
"options": {
"printWidth": 77,
"printWidth": 77,
"parser": "typescript"
}
},

View File

@ -17,7 +17,7 @@ export default tseslint.config(
},
js.configs.recommended,
...tseslint.configs.strictTypeChecked,
...tseslint.configs.stylisticTypeChecked,
@ -49,10 +49,10 @@ export default tseslint.config(
rules: {
'@typescript-eslint/no-unused-vars': ['error', { argsIgnorePattern: '^_', varsIgnorePattern: '^_', caughtErrorsIgnorePattern: '^_' }],
'@typescript-eslint/consistent-type-imports': ['error', { prefer: 'type-imports', fixStyle: 'inline-type-imports' }],
'@typescript-eslint/explicit-function-return-type': ['error', {
allowExpressions: true,
allowTypedFunctionExpressions: true,
allowHigherOrderFunctions: true,
'@typescript-eslint/explicit-function-return-type': ['error', {
allowExpressions: true,
allowTypedFunctionExpressions: true,
allowHigherOrderFunctions: true,
allowDirectConstAssertionInArrowFunctions: true,
allowedNames: ['Component']
}],
@ -60,9 +60,9 @@ export default tseslint.config(
'@typescript-eslint/no-non-null-assertion': 'error',
'@typescript-eslint/array-type': ['error', { default: 'array' }],
'@typescript-eslint/no-explicit-any': 'error',
'@typescript-eslint/no-confusing-void-expression': 'off',
'@typescript-eslint/no-unnecessary-condition': 'off',
'@typescript-eslint/no-floating-promises': 'error',
'@typescript-eslint/no-confusing-void-expression': 'off',
'@typescript-eslint/no-unnecessary-condition': 'off',
'@typescript-eslint/no-floating-promises': 'error',
'@typescript-eslint/strict-boolean-expressions': ['error', {
allowString: false,
allowNumber: false,

View File

@ -1 +1 @@
{"name":"","short_name":"","icons":[{"src":"/android-chrome-192x192.png","sizes":"192x192","type":"image/png"},{"src":"/android-chrome-512x512.png","sizes":"512x512","type":"image/png"}],"theme_color":"#ffffff","background_color":"#ffffff","display":"standalone"}
{"name":"","short_name":"","icons":[{"src":"/android-chrome-192x192.png","sizes":"192x192","type":"image/png"},{"src":"/android-chrome-512x512.png","sizes":"512x512","type":"image/png"}],"theme_color":"#ffffff","background_color":"#ffffff","display":"standalone"}

View File

@ -15,7 +15,7 @@
"isolatedModules": true,
"baseUrl": ".",
"paths": {
"@/*": ["./src/*"],
"@/*": ["./src/*"]
},
"strict": true,
"noUnusedLocals": true,

View File

@ -28,6 +28,6 @@
"react",
"docker"
],
"author": "CarerPerez-dev - also replace my github url username with yours",
"author": "CarerPerez-dev - also replace my github url username with yours",
"license": "MIT"
}

View File

@ -3,7 +3,7 @@ CarterPerez-dev | 2025
This keylogger demonstrates:
keyboard event capture, log management, and remote delivery
Unauthorized use of keyloggers is illegal.
Unauthorized use of keyloggers is illegal.
Only use on systems you own or have permission to monitor
"""
@ -11,11 +11,11 @@ import sys
import logging
import platform
from enum import (
Enum,
Enum,
auto,
)
from threading import (
Event,
Event,
Lock,
)
from pathlib import Path

View File

@ -822,5 +822,3 @@ A collection of tools, courses, frameworks, and educational resources for cyber
- Use practice exams to measure progress
- Join communities for support and networking
- Stay current with security news and trends