Owasp Mobile Security Checker

#!/usr/bin/env python3
"""
Analyze data storage security in Flutter project (OWASP M9).

This script checks for insecure data storage patterns, including:
- Unencrypted SharedPreferences usage
- Plaintext file storage
- Insecure database implementations
"""

import re
import sys
import json
from pathlib import Path
from typing import List, Dict


def scan_shared_preferences_usage(file_path: Path) -> List[Dict]:
    """Scan for insecure SharedPreferences usage."""
    findings = []

    try:
        with open(file_path, 'r', encoding='utf-8', errors='ignore') as f:
            content = f.read()
            lines = content.split('\n')

            # Check for SharedPreferences import
            has_shared_prefs = 'package:shared_preferences' in content
            has_secure_storage = 'package:flutter_secure_storage' in content

            if has_shared_prefs:
                # Look for sensitive data being stored
                sensitive_patterns = [
                    (r'setString\(["\'](?:token|auth|password|secret|key|credential)["\']', 'Token/Credential storage'),
                    (r'setString\(["\'].*(?:api|jwt|bearer).*["\']', 'API key storage'),
                    (r'setInt\(["\'](?:pin|otp|code)["\']', 'PIN/OTP storage'),
                ]

                for line_num, line in enumerate(lines, 1):
                    for pattern, issue_type in sensitive_patterns:
                        if re.search(pattern, line, re.IGNORECASE):
                            findings.append({
                                'file': str(file_path),
                                'line': line_num,
                                'issue': f'{issue_type} in unencrypted SharedPreferences',
                                'code': line.strip(),
                                'severity': 'HIGH',
                                'recommendation': 'Use flutter_secure_storage instead'
                            })

    except Exception as e:
        print(f"Error scanning {file_path}: {e}")

    return findings


def scan_file_storage(file_path: Path) -> List[Dict]:
    """Scan for insecure file storage patterns."""
    findings = []

    try:
        with open(file_path, 'r', encoding='utf-8', errors='ignore') as f:
            content = f.read()
            lines = content.split('\n')

            # Patterns for insecure file operations
            insecure_patterns = [
                (r'File\(.*\)\.writeAsString\((?!.*encrypt)', 'Unencrypted file write'),
                (r'File\(.*\)\.writeAsBytes\((?!.*encrypt)', 'Unencrypted file write'),
                (r'openWrite\(', 'Potentially unencrypted stream write'),
            ]

            for line_num, line in enumerate(lines, 1):
                for pattern, issue_type in insecure_patterns:
                    if re.search(pattern, line):
                        # Check if encryption is mentioned nearby
                        context_start = max(0, line_num - 3)
                        context_end = min(len(lines), line_num + 3)
                        context = '\n'.join(lines[context_start:context_end])

                        if 'encrypt' not in context.lower():
                            findings.append({
                                'file': str(file_path),
                                'line': line_num,
                                'issue': issue_type,
                                'code': line.strip(),
                                'severity': 'MEDIUM',
                                'recommendation': 'Encrypt sensitive data before writing to files'
                            })

    except Exception as e:
        print(f"Error scanning {file_path}: {e}")

    return findings


def scan_database_security(file_path: Path) -> List[Dict]:
    """Scan for insecure database implementations."""
    findings = []

    try:
        with open(file_path, 'r', encoding='utf-8', errors='ignore') as f:
            content = f.read()
            lines = content.split('\n')

            # Check for sqflite without encryption
            has_sqflite = 'package:sqflite' in content
            has_encrypted_sqflite = 'sqflite_sqlcipher' in content

            if has_sqflite and not has_encrypted_sqflite:
                findings.append({
                    'file': str(file_path),
                    'line': 1,
                    'issue': 'Unencrypted SQLite database',
                    'code': 'import package:sqflite',
                    'severity': 'HIGH',
                    'recommendation': 'Consider using sqflite_sqlcipher for encrypted databases'
                })

            # Check for raw SQL queries (injection risk)
            for line_num, line in enumerate(lines, 1):
                if re.search(r'rawQuery\(["\']SELECT.*\$', line):
                    findings.append({
                        'file': str(file_path),
                        'line': line_num,
                        'issue': 'Potential SQL injection via string interpolation',
                        'code': line.strip(),
                        'severity': 'HIGH',
                        'recommendation': 'Use parameterized queries with whereArgs'
                    })

    except Exception as e:
        print(f"Error scanning {file_path}: {e}")

    return findings


def scan_backup_configuration(project_root: Path) -> List[Dict]:
    """Check Android backup configuration."""
    findings = []

    android_manifest = project_root / 'android' / 'app' / 'src' / 'main' / 'AndroidManifest.xml'

    if android_manifest.exists():
        try:
            with open(android_manifest, 'r') as f:
                content = f.read()

                # Check if backup is disabled or restricted
                if 'android:allowBackup="false"' not in content and 'android:fullBackupContent' not in content:
                    findings.append({
                        'file': str(android_manifest),
                        'line': 0,
                        'issue': 'Backup not explicitly configured',
                        'code': '<application>',
                        'severity': 'MEDIUM',
                        'recommendation': 'Set android:allowBackup="false" or configure android:fullBackupContent'
                    })

                if 'android:allowBackup="true"' in content:
                    findings.append({
                        'file': str(android_manifest),
                        'line': 0,
                        'issue': 'Full backup enabled - sensitive data may be backed up',
                        'code': 'android:allowBackup="true"',
                        'severity': 'HIGH',
                        'recommendation': 'Disable backup or use selective backup rules'
                    })

        except Exception as e:
            print(f"Error scanning {android_manifest}: {e}")

    return findings


def scan_project(project_root: str) -> Dict:
    """Scan entire Flutter project for storage security issues."""
    project_path = Path(project_root)
    all_findings = []

    # Get all Dart files
    dart_files = list(project_path.glob('lib/**/*.dart'))

    print(f"Scanning {len(dart_files)} Dart files for storage security issues...")

    for file_path in dart_files:
        findings = []
        findings.extend(scan_shared_preferences_usage(file_path))
        findings.extend(scan_file_storage(file_path))
        findings.extend(scan_database_security(file_path))
        all_findings.extend(findings)

    # Check Android backup configuration
    backup_findings = scan_backup_configuration(project_path)
    all_findings.extend(backup_findings)

    return {
        'total_files_scanned': len(dart_files),
        'total_findings': len(all_findings),
        'findings': all_findings
    }


def main():
    """Main execution function."""
    project_root = sys.argv[1] if len(sys.argv) > 1 else '.'

    print(f"OWASP M9: Analyzing data storage security in {project_root}\n")

    results = scan_project(project_root)

    print(f"\n{'='*60}")
    print("Storage Security Scan Results:")
    print(f"{'='*60}")
    print(f"Files scanned: {results['total_files_scanned']}")
    print(f"Issues found: {results['total_findings']}\n")

    if results['findings']:
        # Group by severity
        by_severity = {'CRITICAL': [], 'HIGH': [], 'MEDIUM': [], 'LOW': []}
        for finding in results['findings']:
            severity = finding.get('severity', 'MEDIUM')
            by_severity[severity].append(finding)

        for severity in ['CRITICAL', 'HIGH', 'MEDIUM', 'LOW']:
            if by_severity[severity]:
                print(f"\n{severity} Severity ({len(by_severity[severity])}):")
                print('-' * 60)
                for i, finding in enumerate(by_severity[severity], 1):
                    print(f"\n{i}. {finding['issue']}")
                    print(f"   File: {finding['file']}:{finding['line']}")
                    if finding.get('code'):
                        print(f"   Code: {finding['code']}")
                    print(f"   Recommendation: {finding['recommendation']}")

    # Save results
    output_file = Path(project_root) / 'owasp_m9_storage_scan.json'
    with open(output_file, 'w') as f:
        json.dump(results, f, indent=2)

    print(f"\n{'='*60}")
    print(f"Results saved to: {output_file}")

    # Exit with error only on CRITICAL/HIGH (consistent with other scanners)
    critical_high = sum(
        1 for f in results['findings'] if f.get('severity') in ('CRITICAL', 'HIGH')
    )
    sys.exit(1 if critical_high > 0 else 0)


if __name__ == '__main__':
    main()

#!/usr/bin/env python3
"""
Check Flutter dependencies for security vulnerabilities (OWASP M2).

This script analyzes pubspec.yaml for outdated packages, version constraints,
and known vulnerabilities.
"""

import re
import yaml
import json
import shutil
import subprocess
import sys
from pathlib import Path
from typing import Dict, List, Tuple


def load_pubspec(project_root: str) -> Dict:
    """Load and parse pubspec.yaml."""
    pubspec_path = Path(project_root) / 'pubspec.yaml'

    if not pubspec_path.exists():
        print(f"Error: pubspec.yaml not found in {project_root}")
        sys.exit(1)

    with open(pubspec_path, 'r') as f:
        return yaml.safe_load(f)


def check_version_constraints(pubspec: Dict) -> List[Dict]:
    """Check for insecure version constraints."""
    findings = []

    dependencies = pubspec.get('dependencies', {})
    dev_dependencies = pubspec.get('dev_dependencies', {})

    all_deps = {**dependencies, **dev_dependencies}

    for package, version in all_deps.items():
        if package == 'flutter':
            continue

        # Check for 'any' version constraint (highly insecure)
        if version == 'any':
            findings.append({
                'package': package,
                'issue': 'Version constraint set to "any"',
                'severity': 'CRITICAL',
                'recommendation': 'Use specific version constraints (e.g., ^1.0.0)'
            })

        # Check for missing version constraint
        elif version is None:
            findings.append({
                'package': package,
                'issue': 'No version constraint specified',
                'severity': 'HIGH',
                'recommendation': 'Specify a version constraint'
            })

    return findings


def check_outdated_packages(project_root: str) -> Dict:
    """Run 'flutter pub outdated' to check for outdated packages."""
    try:
        # Resolve fvm to its absolute path to avoid PATH-insertion attacks
        fvm_path = shutil.which('fvm')
        if fvm_path is not None:
            flutter_cmd = [fvm_path, 'flutter', 'pub', 'outdated', '--json']
        else:
            flutter_cmd = ['flutter', 'pub', 'outdated', '--json']

        result = subprocess.run(
            flutter_cmd,
            cwd=project_root,
            capture_output=True,
            text=True,
            timeout=60
        )

        if result.returncode == 0:
            return json.loads(result.stdout)
        else:
            print(f"Warning: flutter pub outdated failed: {result.stderr}")
            return {}
    except subprocess.TimeoutExpired:
        print("Warning: flutter pub outdated timed out")
        return {}
    except FileNotFoundError:
        print("Warning: Flutter command not found (tried 'fvm flutter' and 'flutter'). Skipping outdated check.")
        return {}
    except Exception as e:
        print(f"Warning: Error running flutter pub outdated: {e}")
        return {}


def analyze_outdated_results(outdated_data: Dict) -> List[Dict]:
    """Analyze outdated packages data."""
    findings = []

    packages = outdated_data.get('packages', [])

    for package in packages:
        current_info = package.get('current')
        latest_info = package.get('latest')
        resolvable_info = package.get('resolvable')

        current = current_info.get('version') if isinstance(current_info, dict) else None
        latest = latest_info.get('version') if isinstance(latest_info, dict) else None
        resolvable = resolvable_info.get('version') if isinstance(resolvable_info, dict) else None

        package_name = package.get('package')

        if current != latest:
            severity = 'MEDIUM'
            # Determine severity based on version gap
            if current and latest:
                current_major = int(current.split('.')[0]) if current.split('.')[0].isdigit() else 0
                latest_major = int(latest.split('.')[0]) if latest.split('.')[0].isdigit() else 0

                if latest_major > current_major:
                    severity = 'HIGH'

            findings.append({
                'package': package_name,
                'current_version': current,
                'latest_version': latest,
                'resolvable_version': resolvable,
                'severity': severity,
                'issue': 'Package is outdated',
                'recommendation': f'Update to version {latest}'
            })

    return findings


def parse_min_version(constraint) -> Tuple[int, int, int]:
    """Extract the minimum pinned version from a pubspec constraint string."""
    if constraint is None or str(constraint).strip() == 'any':
        return (0, 0, 0)
    match = re.search(r'(\d+)\.(\d+)\.(\d+)', str(constraint))
    if match:
        return (int(match.group(1)), int(match.group(2)), int(match.group(3)))
    return (0, 0, 0)


def version_below_threshold(version: Tuple[int, int, int], threshold_str: str) -> bool:
    """Return True if version is below the threshold expressed as '<X.Y.Z'."""
    match = re.search(r'(\d+)\.(\d+)\.(\d+)', threshold_str)
    if not match:
        return False
    threshold = (int(match.group(1)), int(match.group(2)), int(match.group(3)))
    return version < threshold


def check_dangerous_packages(pubspec: Dict) -> List[Dict]:
    """Check for known packages where old versions have security issues."""
    findings = []

    dangerous_packages = {
        'http': {
            'versions': ['<0.13.0'],
            'issue': 'Old versions lack important security features',
            'severity': 'HIGH',
        },
        'path_provider': {
            'versions': ['<2.0.0'],
            'issue': 'Older versions have path traversal vulnerabilities',
            'severity': 'MEDIUM',
        },
    }

    dependencies = pubspec.get('dependencies', {})

    for package, version_info in dangerous_packages.items():
        if package not in dependencies:
            continue
        min_version = parse_min_version(dependencies[package])
        for threshold in version_info['versions']:
            if version_below_threshold(min_version, threshold):
                findings.append({
                    'package': package,
                    'current_constraint': str(dependencies[package]),
                    'issue': version_info['issue'],
                    'severity': version_info['severity'],
                    'recommendation': f'Update to a version that satisfies {threshold.replace("<", ">=")}',
                })

    return findings


def main():
    """Main execution function."""
    project_root = sys.argv[1] if len(sys.argv) > 1 else '.'

    print(f"OWASP M2: Analyzing dependencies in {project_root}\n")

    # Load pubspec.yaml
    pubspec = load_pubspec(project_root)

    all_findings = []

    # Check version constraints
    print("Checking version constraints...")
    constraint_findings = check_version_constraints(pubspec)
    all_findings.extend(constraint_findings)

    # Check for outdated packages
    print("Checking for outdated packages...")
    outdated_data = check_outdated_packages(project_root)
    if outdated_data:
        outdated_findings = analyze_outdated_results(outdated_data)
        all_findings.extend(outdated_findings)

    # Check for dangerous packages
    print("Checking for known vulnerable packages...")
    dangerous_findings = check_dangerous_packages(pubspec)
    all_findings.extend(dangerous_findings)

    # Group by severity (initialised before the findings check so it's always defined)
    by_severity: Dict[str, List] = {'CRITICAL': [], 'HIGH': [], 'MEDIUM': [], 'LOW': []}
    for finding in all_findings:
        severity = finding.get('severity', 'MEDIUM')
        by_severity.setdefault(severity, []).append(finding)

    # Print results
    print(f"\n{'='*60}")
    print("Dependency Scan Results:")
    print(f"{'='*60}")
    print(f"Total issues found: {len(all_findings)}\n")

    for severity in ['CRITICAL', 'HIGH', 'MEDIUM', 'LOW']:
        if by_severity[severity]:
            print(f"\n{severity} Severity ({len(by_severity[severity])}):")
            print('-' * 60)
            for finding in by_severity[severity]:
                print(f"\nPackage: {finding['package']}")
                print(f"Issue: {finding['issue']}")
                if 'current_version' in finding:
                    print(f"Current: {finding['current_version']}, Latest: {finding['latest_version']}")
                print(f"Recommendation: {finding['recommendation']}")

    # Save results
    output_file = Path(project_root) / 'owasp_m2_dependencies_scan.json'
    with open(output_file, 'w') as f:
        json.dump({
            'total_issues': len(all_findings),
            'findings': all_findings
        }, f, indent=2)

    print(f"\n{'='*60}")
    print(f"Results saved to: {output_file}")

    critical_high = len(by_severity['CRITICAL']) + len(by_severity['HIGH'])
    sys.exit(1 if critical_high > 0 else 0)


if __name__ == '__main__':
    main()

#!/usr/bin/env python3
"""
Check network security configurations (OWASP M5).

This script analyzes Flutter project for:
- HTTP vs HTTPS usage
- Certificate pinning implementation
- Network security configuration
- WebSocket security
"""

import re
import sys
import json
import xml.etree.ElementTree as ET
from pathlib import Path
from typing import List, Dict


def scan_http_usage(file_path: Path) -> List[Dict]:
    """Scan for insecure HTTP usage instead of HTTPS."""
    findings = []

    http_patterns = [
        (r'http://(?!localhost|127\.0\.0\.1|0\.0\.0\.0)', 'HTTP URL (non-localhost)'),
        (r'["\']http:["\']', 'HTTP scheme'),
        (r'ws://', 'Insecure WebSocket'),
    ]

    try:
        with open(file_path, 'r', encoding='utf-8', errors='ignore') as f:
            lines = f.read().split('\n')

        in_block_comment = False
        for line_num, line in enumerate(lines, 1):
            stripped = line.strip()

            if in_block_comment:
                if '*/' in line:
                    in_block_comment = False
                continue
            if stripped.startswith('/*'):
                if '*/' not in line:
                    in_block_comment = True
                continue
            if stripped.startswith('//'):
                continue

            for pattern, issue_type in http_patterns:
                for _ in re.finditer(pattern, line, re.IGNORECASE):
                    findings.append({
                        'file': str(file_path),
                        'line': line_num,
                        'issue': issue_type,
                        'code': stripped,
                        'severity': 'HIGH',
                        'recommendation': 'Use HTTPS/WSS instead of HTTP/WS',
                    })

    except Exception as e:
        print(f"Error scanning {file_path}: {e}")

    return findings


def check_bad_cert_callback(file_path: Path) -> List[Dict]:
    """Check for dangerous badCertificateCallback that accepts all certificates."""
    findings = []
    try:
        with open(file_path, 'r', encoding='utf-8', errors='ignore') as f:
            content = f.read()
        if re.search(r'badCertificateCallback.*=.*\(.*\).*=>.*true', content):
            findings.append({
                'file': str(file_path),
                'line': 0,
                'issue': 'Certificate validation disabled (accepts all certificates)',
                'code': 'badCertificateCallback = (...) => true',
                'severity': 'CRITICAL',
                'recommendation': 'Remove this in production or implement proper certificate pinning',
            })
    except Exception as e:
        print(f"Error scanning {file_path}: {e}")
    return findings


def check_http_client_pinning(project_root: Path, dart_files: List[Path]) -> List[Dict]:
    """Project-level check: report once if HttpClient is used without a pinning package."""
    uses_http_client = False
    for file_path in dart_files:
        try:
            with open(file_path, 'r', encoding='utf-8', errors='ignore') as f:
                if 'HttpClient' in f.read():
                    uses_http_client = True
                    break
        except Exception:
            pass

    if not uses_http_client:
        return []

    pubspec_path = project_root / 'pubspec.yaml'
    has_pinning_package = False
    if pubspec_path.exists():
        try:
            content = pubspec_path.read_text()
            has_pinning_package = any(pkg in content for pkg in [
                'http_certificate_pinning',
                'cert_pinning',
                'ssl_pinning_plugin',
            ])
        except Exception:
            pass

    if has_pinning_package:
        return []

    return [{
        'file': 'pubspec.yaml',
        'line': 0,
        'issue': 'HttpClient used without a certificate pinning package',
        'code': 'No http_certificate_pinning / ssl_pinning_plugin found in pubspec.yaml',
        'severity': 'MEDIUM',
        'recommendation': 'Consider implementing certificate pinning for sensitive APIs',
    }]


def check_android_network_security(project_root: Path) -> List[Dict]:
    """Check Android network security configuration."""
    findings = []

    network_security_config = project_root / 'android' / 'app' / 'src' / 'main' / 'res' / 'xml' / 'network_security_config.xml'

    if not network_security_config.exists():
        findings.append({
            'file': 'android/app/src/main/res/xml/network_security_config.xml',
            'line': 0,
            'issue': 'Network security configuration not found',
            'code': 'Missing file',
            'severity': 'MEDIUM',
            'recommendation': 'Create network_security_config.xml to enforce HTTPS and certificate pinning'
        })
        return findings

    try:
        tree = ET.parse(network_security_config)
        root = tree.getroot()

        # Check for cleartextTrafficPermitted
        base_config = root.find('.//base-config')
        if base_config is not None:
            cleartext = base_config.get('cleartextTrafficPermitted')
            if cleartext == 'true':
                findings.append({
                    'file': str(network_security_config),
                    'line': 0,
                    'issue': 'Cleartext traffic permitted globally',
                    'code': 'cleartextTrafficPermitted="true"',
                    'severity': 'HIGH',
                    'recommendation': 'Disable cleartext traffic or limit to specific domains'
                })

        # Check for certificate pinning
        has_pinning = root.findall('.//pin-set')
        if not has_pinning:
            findings.append({
                'file': str(network_security_config),
                'line': 0,
                'issue': 'No certificate pinning configured',
                'code': 'No <pin-set> elements found',
                'severity': 'MEDIUM',
                'recommendation': 'Configure certificate pinning for production APIs'
            })

    except Exception as e:
        print(f"Error parsing network security config: {e}")

    return findings


def check_ios_ats_configuration(project_root: Path) -> List[Dict]:
    """Check iOS App Transport Security configuration."""
    findings = []

    info_plist = project_root / 'ios' / 'Runner' / 'Info.plist'

    if not info_plist.exists():
        return findings

    try:
        with open(info_plist, 'r') as f:
            content = f.read()

            # Check if ATS is disabled globally
            if 'NSAllowsArbitraryLoads' in content:
                # Extract the value
                if re.search(r'NSAllowsArbitraryLoads.*<true/>', content, re.DOTALL):
                    findings.append({
                        'file': str(info_plist),
                        'line': 0,
                        'issue': 'App Transport Security disabled globally',
                        'code': 'NSAllowsArbitraryLoads = true',
                        'severity': 'CRITICAL',
                        'recommendation': 'Enable ATS and use NSExceptionDomains for specific cases only'
                    })

            # Check for local networking allowance
            if 'NSAllowsLocalNetworking' in content:
                findings.append({
                    'file': str(info_plist),
                    'line': 0,
                    'issue': 'Local networking allowed',
                    'code': 'NSAllowsLocalNetworking',
                    'severity': 'LOW',
                    'recommendation': 'Ensure this is intentional and only for development'
                })

    except Exception as e:
        print(f"Error scanning Info.plist: {e}")

    return findings


def scan_project(project_root: str) -> Dict:
    """Scan entire Flutter project for network security issues."""
    project_path = Path(project_root)
    all_findings = []

    # Get all Dart files
    dart_files = list(project_path.glob('lib/**/*.dart'))

    print(f"Scanning {len(dart_files)} Dart files for network security issues...")

    for file_path in dart_files:
        all_findings.extend(scan_http_usage(file_path))
        all_findings.extend(check_bad_cert_callback(file_path))

    # Project-level pinning check — emits one finding, not one per file
    all_findings.extend(check_http_client_pinning(project_path, dart_files))

    all_findings.extend(check_android_network_security(project_path))
    all_findings.extend(check_ios_ats_configuration(project_path))

    return {
        'total_files_scanned': len(dart_files),
        'total_findings': len(all_findings),
        'findings': all_findings
    }


def main():
    """Main execution function."""
    project_root = sys.argv[1] if len(sys.argv) > 1 else '.'

    print(f"OWASP M5: Analyzing network security in {project_root}\n")

    results = scan_project(project_root)

    # Group by severity (always initialised so the exit-code line is always safe)
    by_severity: Dict[str, List] = {'CRITICAL': [], 'HIGH': [], 'MEDIUM': [], 'LOW': []}
    for finding in results['findings']:
        by_severity.setdefault(finding.get('severity', 'MEDIUM'), []).append(finding)

    print(f"\n{'='*60}")
    print("Network Security Scan Results:")
    print(f"{'='*60}")
    print(f"Files scanned: {results['total_files_scanned']}")
    print(f"Issues found: {results['total_findings']}\n")

    for severity in ['CRITICAL', 'HIGH', 'MEDIUM', 'LOW']:
        if by_severity[severity]:
            print(f"\n{severity} Severity ({len(by_severity[severity])}):")
            print('-' * 60)
            for i, finding in enumerate(by_severity[severity], 1):
                print(f"\n{i}. {finding['issue']}")
                print(f"   File: {finding['file']}:{finding['line']}")
                if finding.get('code'):
                    print(f"   Code: {finding['code']}")
                print(f"   Recommendation: {finding['recommendation']}")

    # Save results
    output_file = Path(project_root) / 'owasp_m5_network_scan.json'
    with open(output_file, 'w') as f:
        json.dump(results, f, indent=2)

    print(f"\n{'='*60}")
    print(f"Results saved to: {output_file}")

    critical_high = len(by_severity['CRITICAL']) + len(by_severity['HIGH'])
    sys.exit(1 if critical_high > 0 else 0)


if __name__ == '__main__':
    main()

#!/usr/bin/env python3
"""
Scan Flutter project for hardcoded secrets and credentials (OWASP M1).

This script searches for common patterns of hardcoded API keys, tokens,
passwords, and other sensitive credentials in Dart source files.
"""

import re
import os
import sys
import json
from pathlib import Path
from typing import List, Dict, Tuple

# Patterns for detecting hardcoded secrets — (regex, label, severity)
SECRET_PATTERNS = [
    # API Keys and Tokens
    (r'api[_-]?key\s*[:=]\s*["\']([a-zA-Z0-9_\-]{20,})["\']', 'API Key', 'HIGH'),
    (r'apikey\s*[:=]\s*["\']([a-zA-Z0-9_\-]{20,})["\']', 'API Key', 'HIGH'),
    (r'api[_-]?secret\s*[:=]\s*["\']([a-zA-Z0-9_\-]{20,})["\']', 'API Secret', 'HIGH'),
    (r'access[_-]?token\s*[:=]\s*["\']([a-zA-Z0-9_\-]{20,})["\']', 'Access Token', 'HIGH'),
    (r'auth[_-]?token\s*[:=]\s*["\']([a-zA-Z0-9_\-]{20,})["\']', 'Auth Token', 'HIGH'),
    (r'bearer\s+["\']([a-zA-Z0-9_\-\.]{20,})["\']', 'Bearer Token', 'HIGH'),

    # AWS Credentials
    (r'aws[_-]?access[_-]?key[_-]?id\s*[:=]\s*["\']([A-Z0-9]{20})["\']', 'AWS Access Key', 'CRITICAL'),
    (r'aws[_-]?secret[_-]?access[_-]?key\s*[:=]\s*["\']([a-zA-Z0-9/+=]{40})["\']', 'AWS Secret Key', 'CRITICAL'),

    # Firebase
    (r'firebase[_-]?api[_-]?key\s*[:=]\s*["\']([a-zA-Z0-9_\-]{30,})["\']', 'Firebase API Key', 'HIGH'),

    # Database credentials
    (r'db[_-]?password\s*[:=]\s*["\']([^"\']+)["\']', 'Database Password', 'HIGH'),
    (r'database[_-]?url\s*[:=]\s*["\'][^"\']*:[^"\']*@[^"\']+["\']', 'Database URL with credentials', 'CRITICAL'),

    # Generic passwords
    (r'password\s*[:=]\s*["\']([^"\']{8,})["\']', 'Password', 'MEDIUM'),
    (r'passwd\s*[:=]\s*["\']([^"\']{8,})["\']', 'Password', 'MEDIUM'),
    (r'pwd\s*[:=]\s*["\']([^"\']{8,})["\']', 'Password', 'MEDIUM'),

    # Private keys
    (r'private[_-]?key\s*[:=]\s*["\']([^"\']+)["\']', 'Private Key', 'CRITICAL'),
    (r'-----BEGIN (?:RSA |DSA )?PRIVATE KEY-----', 'Private Key Block', 'CRITICAL'),

    # OAuth and Client Secrets
    (r'client[_-]?secret\s*[:=]\s*["\']([a-zA-Z0-9_\-]{20,})["\']', 'Client Secret', 'HIGH'),
    (r'client[_-]?id\s*[:=]\s*["\']([a-zA-Z0-9_\-]{20,})["\']', 'Client ID', 'MEDIUM'),

    # Generic secrets
    (r'secret[_-]?key\s*[:=]\s*["\']([a-zA-Z0-9_\-]{16,})["\']', 'Secret Key', 'HIGH'),
    (r'encryption[_-]?key\s*[:=]\s*["\']([a-zA-Z0-9_\-/+=]{16,})["\']', 'Encryption Key', 'HIGH'),
]

# False positive patterns to exclude
FALSE_POSITIVE_PATTERNS = [
    r'example\.com',
    r'your[_-]api[_-]key',
    r'YOUR[_-]API[_-]KEY',
    r'<.*>',  # Placeholder patterns
    r'\$\{.*\}',  # Template variables
    r'FIXME',
    r'TODO',
    r'dummy',
    r'\btest\b',
    r'sample',
    r'placeholder',
]


def is_false_positive(value: str) -> bool:
    """Check if the detected value is likely a false positive."""
    value_lower = value.lower()
    for pattern in FALSE_POSITIVE_PATTERNS:
        if re.search(pattern, value_lower, re.IGNORECASE):
            return True
    return False


def scan_file(file_path: Path) -> List[Dict]:
    """Scan a single file for hardcoded secrets."""
    findings = []

    try:
        with open(file_path, 'r', encoding='utf-8', errors='ignore') as f:
            content = f.read()
            lines = content.split('\n')

            in_block_comment = False
            for line_num, line in enumerate(lines, 1):
                stripped = line.strip()

                # Track block comments (Dart does not support nested /* */)
                if in_block_comment:
                    if '*/' in line:
                        in_block_comment = False
                    continue
                if stripped.startswith('/*'):
                    if '*/' not in line:
                        in_block_comment = True
                    continue

                # Skip single-line comments
                if stripped.startswith('//'):
                    continue

                for pattern, secret_type, severity in SECRET_PATTERNS:
                    matches = re.finditer(pattern, line, re.IGNORECASE)
                    for match in matches:
                        value = match.group(1) if len(match.groups()) > 0 else match.group(0)

                        if is_false_positive(value):
                            continue

                        findings.append({
                            'file': str(file_path),
                            'line': line_num,
                            'type': secret_type,
                            'pattern': line.strip(),
                            'severity': severity,
                        })
    except Exception as e:
        print(f"Error scanning {file_path}: {e}")

    return findings


def scan_project(project_root: str) -> Dict:
    """Scan entire Flutter project for hardcoded secrets."""
    project_path = Path(project_root)
    all_findings = []

    dart_files = list(project_path.glob('lib/**/*.dart'))
    gradle_files = list(project_path.glob('android/**/*.gradle'))
    plist_files = list(project_path.glob('ios/**/*.plist'))

    files_to_scan = dart_files + gradle_files + plist_files

    print(f"Scanning {len(files_to_scan)} files for hardcoded secrets...")

    for file_path in files_to_scan:
        findings = scan_file(file_path)
        all_findings.extend(findings)

    return {
        'total_files_scanned': len(files_to_scan),
        'total_findings': len(all_findings),
        'findings': all_findings
    }


def main():
    """Main execution function."""
    if len(sys.argv) < 2:
        project_root = os.getcwd()
    else:
        project_root = sys.argv[1]

    print(f"OWASP M1: Scanning for hardcoded secrets in {project_root}\n")

    results = scan_project(project_root)

    print(f"\n{'='*60}")
    print(f"Scan Results:")
    print(f"{'='*60}")
    print(f"Files scanned: {results['total_files_scanned']}")
    print(f"Potential secrets found: {results['total_findings']}\n")

    if results['findings']:
        print("Findings by type:")
        findings_by_type = {}
        for finding in results['findings']:
            secret_type = finding['type']
            findings_by_type[secret_type] = findings_by_type.get(secret_type, 0) + 1

        for secret_type, count in sorted(findings_by_type.items()):
            print(f"  - {secret_type}: {count}")

        print(f"\n{'='*60}")
        print("Detailed Findings:")
        print(f"{'='*60}\n")

        for i, finding in enumerate(results['findings'], 1):
            print(f"{i}. [{finding['severity']}] {finding['type']}")
            print(f"   File: {finding['file']}:{finding['line']}")
            print(f"   Code: {finding['pattern']}")
            print()

    # Save results to JSON
    output_file = Path(project_root) / 'owasp_m1_secrets_scan.json'
    with open(output_file, 'w') as f:
        json.dump(results, f, indent=2)

    print(f"Results saved to: {output_file}")

    # Exit with error code only on CRITICAL/HIGH findings (consistent with other scanners)
    critical_high = sum(
        1 for f in results['findings'] if f.get('severity') in ('CRITICAL', 'HIGH')
    )
    sys.exit(1 if critical_high > 0 else 0)


if __name__ == '__main__':
    main()