RomanNum3ral
/
Cybersecurity-Projects
mirror of https://github.com/CarterPerez-dev/Cybersecurity-Projects.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

yapf

This commit is contained in:
CarterPerez-dev 2025-11-12 15:33:54 -05:00
parent e7a4ea6f92
commit 116460f01a
20 changed files with 593 additions and 347 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
.pre-commit-config.yaml
View File

@ -9,13 +9,16 @@ repos:
files: ^PROJECTS/api-security-scanner/backend/ files: ^PROJECTS/api-security-scanner/backend/
exclude: ^PROJECTS/api-security-scanner/backend/(\.venv|__pycache__|\.pytest_cache)/ 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 - repo: local
hooks: hooks:
- id: yapf
name: yapf format (backend)
entry: bash -c 'cd PROJECTS/api-security-scanner/backend && yapf -i -r -vv models/ repositories/ schemas/ scanners/ core/ factory/'
language: system
types: [python]
files: ^PROJECTS/api-security-scanner/backend/
pass_filenames: false
- id: mypy - id: mypy
name: mypy type check (backend) name: mypy type check (backend)
entry: bash -c 'cd PROJECTS/api-security-scanner/backend && mypy .' entry: bash -c 'cd PROJECTS/api-security-scanner/backend && mypy .'

10
PROJECTS/api-security-scanner/backend/core/database.py
View File

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

18
PROJECTS/api-security-scanner/backend/core/dependencies.py
View File

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

27
PROJECTS/api-security-scanner/backend/core/security.py
View File

@ -22,7 +22,10 @@ def hash_password(password: str) -> str:
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 Verify a plain text password against a hashed password
""" """
@ -31,7 +34,11 @@ def verify_password(plain_password: str, hashed_password: str) -> bool:
return bcrypt.checkpw(password_bytes, hashed_bytes) 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 Create a JWT access token
""" """
@ -40,10 +47,16 @@ def create_access_token(data: dict[str, str], expires_delta: timedelta | None =
if expires_delta: if expires_delta:
expire = datetime.utcnow() + expires_delta expire = datetime.utcnow() + expires_delta
else: 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}) 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 return encoded_jwt
@ -52,7 +65,11 @@ def decode_token(token: str) -> dict[str, str]:
Decode and verify a JWT token Decode and verify a JWT token
""" """
try: try:
payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM]) payload = jwt.decode(
token,
settings.SECRET_KEY,
algorithms = [settings.ALGORITHM]
)
return payload return payload
except JWTError as e: except JWTError as e:
raise ValueError(f"Invalid token: {str(e)}") from e raise ValueError(f"Invalid token: {str(e)}") from e

24
PROJECTS/api-security-scanner/backend/models/Base.py
View File

@ -24,12 +24,20 @@ class BaseModel(Base):
__abstract__ = True __abstract__ = True
id = Column(Integer, primary_key=True, index=True, autoincrement=True) id = Column(
created_at = Column(DateTime(timezone=True), default=lambda: datetime.now(UTC)) Integer,
primary_key = True,
index = True,
autoincrement = True
)
created_at = Column(
DateTime(timezone = True),
default = lambda: datetime.now(UTC)
)
updated_at = Column( updated_at = Column(
DateTime(timezone=True), DateTime(timezone = True),
default=lambda: datetime.now(UTC), default = lambda: datetime.now(UTC),
onupdate=lambda: datetime.now(UTC), onupdate = lambda: datetime.now(UTC),
) )
@declared_attr @declared_attr
@ -46,7 +54,11 @@ class BaseModel(Base):
Returns: Returns:
dict: Dictionary representation of the model 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: def update(self, **kwargs: Any) -> None:
""" """

31
PROJECTS/api-security-scanner/backend/models/Scan.py
View File

@ -29,25 +29,26 @@ class Scan(BaseModel):
user_id = Column( user_id = Column(
Integer, Integer,
ForeignKey("users.id", ondelete="CASCADE"), ForeignKey("users.id",
nullable=False, ondelete = "CASCADE"),
index=True, nullable = False,
index = True,
) )
target_url = Column( target_url = Column(
String(settings.URL_MAX_LENGTH), String(settings.URL_MAX_LENGTH),
nullable=False, nullable = False,
) )
scan_date = Column( scan_date = Column(
DateTime(timezone=True), DateTime(timezone = True),
default=lambda: datetime.now(UTC), default = lambda: datetime.now(UTC),
nullable=False, nullable = False,
) )
user = relationship("User", backref="scans") user = relationship("User", backref = "scans")
test_results = relationship( test_results = relationship(
"TestResult", "TestResult",
back_populates="scan", back_populates = "scan",
cascade="all, delete-orphan", cascade = "all, delete-orphan",
) )
def __repr__(self) -> str: def __repr__(self) -> str:
@ -64,7 +65,10 @@ class Scan(BaseModel):
Returns: Returns:
bool: True if any test result is vulnerable 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 @property
def vulnerability_count(self) -> int: def vulnerability_count(self) -> int:
@ -74,4 +78,7 @@ class Scan(BaseModel):
Returns: Returns:
int: Number of vulnerable test results 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"
)

31
PROJECTS/api-security-scanner/backend/models/TestResult.py
View File

@ -30,30 +30,35 @@ class TestResult(BaseModel):
scan_id = Column( scan_id = Column(
Integer, Integer,
ForeignKey("scans.id", ondelete="CASCADE"), ForeignKey("scans.id",
nullable=False, ondelete = "CASCADE"),
index=True, nullable = False,
index = True,
) )
test_name = Column( test_name = Column(
Enum(TestType), Enum(TestType),
nullable=False, nullable = False,
index=True, index = True,
) )
status = Column( status = Column(
Enum(ScanStatus), Enum(ScanStatus),
nullable=False, nullable = False,
index=True, index = True,
) )
severity = Column( severity = Column(
Enum(Severity), Enum(Severity),
nullable=False, nullable = False,
index=True, index = True,
)
details = Column(Text, nullable = False)
evidence_json = Column(JSON, nullable = False, default = dict)
recommendations_json = Column(
JSON,
nullable = False,
default = list
) )
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: def __repr__(self) -> str:
""" """

10
PROJECTS/api-security-scanner/backend/models/User.py
View File

@ -22,12 +22,12 @@ class User(BaseModel):
email = Column( email = Column(
String(settings.EMAIL_MAX_LENGTH), String(settings.EMAIL_MAX_LENGTH),
unique=True, unique = True,
nullable=False, nullable = False,
index=True, index = True,
) )
hashed_password = Column(String, nullable=False) hashed_password = Column(String, nullable = False)
is_active = Column(Boolean, default=True, nullable=False) is_active = Column(Boolean, default = True, nullable = False)
def __repr__(self) -> str: def __repr__(self) -> str:
""" """

55
PROJECTS/api-security-scanner/backend/repositories/scan_repository.py
View File

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

59
PROJECTS/api-security-scanner/backend/repositories/test_result_repository.py
View File

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

28
PROJECTS/api-security-scanner/backend/repositories/user_repository.py
View File

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

126
PROJECTS/api-security-scanner/backend/scanners/auth_scanner.py
View File

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

67
PROJECTS/api-security-scanner/backend/scanners/base_scanner.py
View File

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

87
PROJECTS/api-security-scanner/backend/scanners/idor_scanner.py
View File

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

64
PROJECTS/api-security-scanner/backend/scanners/payloads.py
View File

@ -131,13 +131,10 @@ class SQLiPayloads:
list[str]: All SQLi test payloads list[str]: All SQLi test payloads
""" """
return ( return (
cls.BASIC_AUTHENTICATION_BYPASS cls.BASIC_AUTHENTICATION_BYPASS + cls.UNION_BASED +
+ cls.UNION_BASED cls.TIME_BASED_BLIND + cls.BOOLEAN_BASED_BLIND +
+ cls.TIME_BASED_BLIND cls.ERROR_BASED + cls.STACKED_QUERIES +
+ cls.BOOLEAN_BASED_BLIND cls.COMMENT_VARIATIONS
+ cls.ERROR_BASED
+ cls.STACKED_QUERIES
+ cls.COMMENT_VARIATIONS
) )
@classmethod @classmethod
@ -280,9 +277,12 @@ class RateLimitBypassPayloads:
""" """
HEADER_PATTERNS = { HEADER_PATTERNS = {
"limit": r"x-ratelimit-limit|x-rate-limit-limit|ratelimit-limit", "limit":
"remaining": r"x-ratelimit-remaining|x-rate-limit-remaining|ratelimit-remaining", r"x-ratelimit-limit|x-rate-limit-limit|ratelimit-limit",
"reset": r"x-ratelimit-reset|x-rate-limit-reset|ratelimit-reset", "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", "retry_after": r"retry-after",
} }
@ -300,14 +300,30 @@ class RateLimitBypassPayloads:
] ]
HEADER_SPOOFING = [ HEADER_SPOOFING = [
{"X-Forwarded-For": "127.0.0.1"}, {
{"X-Forwarded-For": "8.8.8.8"}, "X-Forwarded-For": "127.0.0.1"
{"X-Real-IP": "127.0.0.1"}, },
{"X-Originating-IP": "127.0.0.1"}, {
{"X-Remote-IP": "127.0.0.1"}, "X-Forwarded-For": "8.8.8.8"
{"X-Client-IP": "127.0.0.1"}, },
{"CF-Connecting-IP": "127.0.0.1"}, {
{"True-Client-IP": "127.0.0.1"}, "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 = [ USER_AGENT_ROTATION = [
@ -433,14 +449,10 @@ class XSSPayloads:
list[str]: All XSS test payloads list[str]: All XSS test payloads
""" """
return ( return (
cls.BASIC_XSS cls.BASIC_XSS + cls.EVENT_HANDLER_XSS + cls.SVG_XSS +
+ cls.EVENT_HANDLER_XSS cls.IFRAME_XSS + cls.ENCODED_XSS +
+ cls.SVG_XSS cls.ATTRIBUTE_BREAKING + cls.FILTER_BYPASS +
+ cls.IFRAME_XSS cls.POLYGLOT_XSS
+ cls.ENCODED_XSS
+ cls.ATTRIBUTE_BREAKING
+ cls.FILTER_BYPASS
+ cls.POLYGLOT_XSS
) )
@classmethod @classmethod

120
PROJECTS/api-security-scanner/backend/scanners/rate_limit_scanner.py
View File

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

130
PROJECTS/api-security-scanner/backend/scanners/sqli_scanner.py
View File

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

16
PROJECTS/api-security-scanner/backend/schemas/scan_schemas.py
View File

@ -23,13 +23,13 @@ class ScanRequest(BaseModel):
Schema for creating a new security scan 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 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( max_requests: int = Field(
default=settings.DEFAULT_MAX_REQUESTS, default = settings.DEFAULT_MAX_REQUESTS,
ge=1, ge = 1,
le=settings.SCANNER_MAX_CONCURRENT_REQUESTS, le = settings.SCANNER_MAX_CONCURRENT_REQUESTS,
) )
@ -38,7 +38,7 @@ class ScanResponse(BaseModel):
Schema for scan data in API responses Schema for scan data in API responses
""" """
model_config = ConfigDict(from_attributes=True) model_config = ConfigDict(from_attributes = True)
id: int id: int
user_id: int user_id: int
@ -59,4 +59,6 @@ class ScanResponse(BaseModel):
""" """
Number of vulnerabilities found 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"
)

6
PROJECTS/api-security-scanner/backend/schemas/test_result_schemas.py
View File

@ -27,8 +27,8 @@ class TestResultCreate(BaseModel):
status: ScanStatus status: ScanStatus
severity: Severity severity: Severity
details: str details: str
evidence_json: dict[str, Any] = Field(default_factory=dict) evidence_json: dict[str, Any] = Field(default_factory = dict)
recommendations_json: list[str] = Field(default_factory=list) recommendations_json: list[str] = Field(default_factory = list)
class TestResultResponse(BaseModel): class TestResultResponse(BaseModel):
@ -36,7 +36,7 @@ class TestResultResponse(BaseModel):
Schema for individual test result in API responses Schema for individual test result in API responses
""" """
model_config = ConfigDict(from_attributes=True) model_config = ConfigDict(from_attributes = True)
id: int id: int
scan_id: int scan_id: int

18
PROJECTS/api-security-scanner/backend/schemas/user_schemas.py
View File

@ -19,8 +19,8 @@ class UserCreate(BaseModel):
email: EmailStr email: EmailStr
password: str = Field( password: str = Field(
min_length=settings.PASSWORD_MIN_LENGTH, min_length = settings.PASSWORD_MIN_LENGTH,
max_length=settings.PASSWORD_MAX_LENGTH, max_length = settings.PASSWORD_MAX_LENGTH,
) )
@field_validator("password") @field_validator("password")
@ -30,11 +30,17 @@ class UserCreate(BaseModel):
Validate password meets security requirements Validate password meets security requirements
""" """
if not re.search(r"[A-Z]", v): if not re.search(r"[A-Z]", v):
raise ValueError("Password must contain at least one uppercase letter") raise ValueError(
"Password must contain at least one uppercase letter"
)
if not re.search(r"[a-z]", v): if not re.search(r"[a-z]", v):
raise ValueError("Password must contain at least one lowercase letter") raise ValueError(
"Password must contain at least one lowercase letter"
)
if not re.search(r"[0-9]", v): if not re.search(r"[0-9]", v):
raise ValueError("Password must contain at least one number") raise ValueError(
"Password must contain at least one number"
)
return v return v
@ -53,7 +59,7 @@ class UserResponse(BaseModel):
Excludes sensitive fields like hashed_password. Excludes sensitive fields like hashed_password.
""" """
model_config = ConfigDict(from_attributes=True) model_config = ConfigDict(from_attributes = True)
id: int id: int
email: str email: str