CalvinBackup/AASRT
1
0
Fork 0
mirror of https://github.com/0xsrb/AASRT.git synced 2026-04-22 17:56:08 +02:00
Code Issues Packages Projects Releases Wiki Activity

fixed security issues

Browse Source
This commit is contained in:
swethab
2026-02-10 15:22:04 -05:00
parent 197aa2d729
commit f5abb069be
7 changed files with 89 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Dockerfile
+2 -2
View File
@@ -76,9 +76,9 @@ USER aasrt
# Expose Streamlit port
EXPOSE 8501
# Health check
# Health check using Python (curl not available in slim image)
HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
CMD curl -f http://localhost:8501/_stcore/health || exit 1
CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:8501/_stcore/health', timeout=5)" || exit 1
# Default command: Run Streamlit web interface
CMD ["streamlit", "run", "app.py", "--server.port=8501", "--server.address=0.0.0.0"]
app.py
+9 -3
View File
@@ -932,7 +932,10 @@ def get_templates():
config = Config()
qm = QueryManager(config)
return sorted(qm.get_available_templates())
except:
except Exception as e:
# Log the error but don't expose details to UI
import logging
logging.getLogger(__name__).warning(f"Failed to load templates: {e}")
return []
@@ -1159,8 +1162,9 @@ def run_scan(
all_results = query_manager.execute_query(query, max_results=max_results)
progress_bar.progress(50)
except Exception as e:
st.error(f"SCAN FAILURE: {e}")
# Security: Log full error details but show sanitized message to user
logger.error(f"Scan execution failed: {e}", exc_info=True)
st.error("SCAN FAILURE: An error occurred during scanning. Please check logs for details.")
progress_container.empty()
return None
@@ -1218,7 +1222,9 @@ def run_scan(
db.add_findings(scan_record.scan_id, unique_results)
db.update_scan(scan_record.scan_id, status='completed', total_results=len(unique_results), duration_seconds=duration)
except Exception as e:
st.warning(f"Database sync failed: {e}")
# Security: Log full error but show sanitized message to user
logger.warning(f"Database sync failed: {e}")
st.warning("Database sync failed. Results are still available but may not be persisted.")
progress_bar.progress(100)
status_text.markdown(f"""
src/enrichment/clawsec_feed.py
+20 -4
View File
@@ -2,6 +2,7 @@
import json
import os
import stat
import threading
from dataclasses import dataclass, field
from datetime import datetime, timedelta
@@ -14,6 +15,9 @@ from src.utils.logger import get_logger
logger = get_logger(__name__)
# Security: Restrictive file permissions for cache files (owner read/write only)
SECURE_FILE_PERMISSIONS = stat.S_IRUSR | stat.S_IWUSR # 0o600
@dataclass
class ClawSecAdvisory:
@@ -54,7 +58,8 @@ class ClawSecAdvisory:
if published and isinstance(published, str):
try:
published = datetime.fromisoformat(published.replace('Z', '+00:00'))
except:
except (ValueError, TypeError) as e:
logger.debug(f"Failed to parse published date: {e}")
published = None
return cls(
@@ -166,7 +171,12 @@ class ClawSecFeedManager:
try:
logger.info(f"Fetching ClawSec feed from {self.feed_url}")
response = requests.get(self.feed_url, timeout=self.timeout)
# Security: Explicit SSL verification to prevent MITM attacks
response = requests.get(
self.feed_url,
timeout=self.timeout,
verify=True # Explicitly verify SSL certificates
)
response.raise_for_status()
data = response.json()
@@ -240,8 +250,14 @@ class ClawSecFeedManager:
'cached_at': datetime.utcnow().isoformat()
}
with open(cache_path, 'w') as f:
json.dump(cache_data, f, indent=2)
# Security: Write with restrictive permissions (owner read/write only)
fd = os.open(str(cache_path), os.O_CREAT | os.O_WRONLY | os.O_TRUNC, SECURE_FILE_PERMISSIONS)
try:
with os.fdopen(fd, 'w') as f:
json.dump(cache_data, f, indent=2)
except Exception:
os.close(fd)
raise
logger.debug(f"ClawSec cache saved to {self.cache_file}")
src/reporting/csv_reporter.py
+29 -6
View File
@@ -2,6 +2,8 @@
import csv
import io
import os
import stat
from typing import List, Optional
from .base import BaseReporter, ScanReport
@@ -9,6 +11,9 @@ from src.utils.logger import get_logger
logger = get_logger(__name__)
# Security: Restrictive file permissions for report files (owner read/write only)
SECURE_FILE_PERMISSIONS = stat.S_IRUSR | stat.S_IWUSR # 0o600
class CSVReporter(BaseReporter):
"""Generates CSV format reports."""
@@ -72,8 +77,14 @@ class CSVReporter(BaseReporter):
content = self.generate_string(report)
with open(filepath, 'w', encoding='utf-8', newline='') as f:
f.write(content)
# Security: Write with restrictive permissions (owner read/write only)
fd = os.open(filepath, os.O_CREAT | os.O_WRONLY | os.O_TRUNC, SECURE_FILE_PERMISSIONS)
try:
with os.fdopen(fd, 'w', encoding='utf-8', newline='') as f:
f.write(content)
except Exception:
os.close(fd)
raise
logger.info(f"Generated CSV report: {filepath}")
return filepath
@@ -171,8 +182,14 @@ class CSVReporter(BaseReporter):
writer.writerow(['Low Findings', report.low_findings])
writer.writerow(['Average Risk Score', report.average_risk_score])
with open(filepath, 'w', encoding='utf-8', newline='') as f:
f.write(output.getvalue())
# Security: Write with restrictive permissions (owner read/write only)
fd = os.open(filepath, os.O_CREAT | os.O_WRONLY | os.O_TRUNC, SECURE_FILE_PERMISSIONS)
try:
with os.fdopen(fd, 'w', encoding='utf-8', newline='') as f:
f.write(output.getvalue())
except Exception:
os.close(fd)
raise
logger.info(f"Generated CSV summary: {filepath}")
return filepath
@@ -214,8 +231,14 @@ class CSVReporter(BaseReporter):
else:
writer.writerow([ip, port, hostname, 'None detected', risk_score])
with open(filepath, 'w', encoding='utf-8', newline='') as f:
f.write(output.getvalue())
# Security: Write with restrictive permissions (owner read/write only)
fd = os.open(filepath, os.O_CREAT | os.O_WRONLY | os.O_TRUNC, SECURE_FILE_PERMISSIONS)
try:
with os.fdopen(fd, 'w', encoding='utf-8', newline='') as f:
f.write(output.getvalue())
except Exception:
os.close(fd)
raise
logger.info(f"Generated vulnerability CSV: {filepath}")
return filepath
src/reporting/json_reporter.py
+13 -2
View File
@@ -1,6 +1,8 @@
"""JSON report generator for AASRT."""
import json
import os
import stat
from typing import Optional
from .base import BaseReporter, ScanReport
@@ -8,6 +10,9 @@ from src.utils.logger import get_logger
logger = get_logger(__name__)
# Security: Restrictive file permissions for report files (owner read/write only)
SECURE_FILE_PERMISSIONS = stat.S_IRUSR | stat.S_IWUSR # 0o600
class JSONReporter(BaseReporter):
"""Generates JSON format reports."""
@@ -47,8 +52,14 @@ class JSONReporter(BaseReporter):
content = self.generate_string(report)
with open(filepath, 'w', encoding='utf-8') as f:
f.write(content)
# Security: Write with restrictive permissions (owner read/write only)
fd = os.open(filepath, os.O_CREAT | os.O_WRONLY | os.O_TRUNC, SECURE_FILE_PERMISSIONS)
try:
with os.fdopen(fd, 'w', encoding='utf-8') as f:
f.write(content)
except Exception:
os.close(fd)
raise
logger.info(f"Generated JSON report: {filepath}")
return filepath
src/storage/database.py
+1 -1
View File
@@ -559,7 +559,7 @@ class Database:
scan = session.query(Scan).filter(Scan.scan_id == scan_id).first()
if scan:
session.delete(scan)
session.commit()
# Note: session_scope() commits automatically on successful exit
logger.info(f"Deleted scan: {scan_id}")
return True
return False
src/utils/validators.py
+15 -2
View File
@@ -242,6 +242,7 @@ def sanitize_output(text: str) -> str:
text = str(text)
# Patterns for sensitive data (order matters - more specific first)
# Security: Context-aware patterns to avoid false positives on legitimate hex strings
patterns = [
# Anthropic API keys
(r'sk-ant-[a-zA-Z0-9-_]{20,}', 'sk-ant-***REDACTED***'),
@@ -259,8 +260,17 @@ def sanitize_output(text: str) -> str:
# Stripe keys
(r'sk_live_[a-zA-Z0-9]{24,}', 'sk_live_***REDACTED***'),
(r'sk_test_[a-zA-Z0-9]{24,}', 'sk_test_***REDACTED***'),
# Shodan API key (32 hex chars)
(r'[a-fA-F0-9]{32}', '***REDACTED_KEY***'),
# Shodan API key with context (more specific patterns first)
# Pattern 1: Shodan key in environment variable or config context
(r'(?i)(?:shodan[_\-]?(?:api)?[_\-]?key|SHODAN_API_KEY)\s*[=:]\s*["\']?([a-fA-F0-9]{32})["\']?',
'SHODAN_API_KEY=***REDACTED***'),
# Pattern 2: API key in JSON/config context
(r'(?i)["\']?(?:api[_\-]?key|apikey)["\']?\s*[=:]\s*["\']?([a-fA-F0-9]{32})["\']?',
'"api_key": "***REDACTED***"'),
# Pattern 3: Standalone 32-char hex that looks like a key (not preceded by common hash prefixes)
# Avoid matching MD5 hashes by checking context - only match if it looks like a credential
(r'(?<![a-fA-F0-9])(?:key|token|secret|credential)["\s:=]+["\']?([a-fA-F0-9]{32})["\']?(?![a-fA-F0-9])',
'***REDACTED_KEY***'),
# Generic password patterns
(r'password["\s:=]+["\']?[\w@#$%^&*!?]+', 'password=***REDACTED***'),
(r'passwd["\s:=]+["\']?[\w@#$%^&*!?]+', 'passwd=***REDACTED***'),
@@ -269,6 +279,9 @@ def sanitize_output(text: str) -> str:
(r'Bearer\s+[a-zA-Z0-9._-]+', 'Bearer ***REDACTED***'),
# Basic auth
(r'Basic\s+[a-zA-Z0-9+/=]+', 'Basic ***REDACTED***'),
# Private keys (PEM format)
(r'-----BEGIN (?:RSA |EC |DSA )?PRIVATE KEY-----[\s\S]*?-----END (?:RSA |EC |DSA )?PRIVATE KEY-----',
'-----BEGIN PRIVATE KEY-----***REDACTED***-----END PRIVATE KEY-----'),
]
result = text