diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..239b0d8 --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,49 @@ +name: CI + +on: + push: + branches: [main] + pull_request: + +jobs: + test: + name: test (${{ matrix.os }}, py${{ matrix.python-version }}) + runs-on: ${{ matrix.os }} + strategy: + fail-fast: false + matrix: + os: [ubuntu-latest, windows-latest, macos-latest] + python-version: ['3.8', '3.9', '3.10', '3.11', '3.12'] + steps: + - uses: actions/checkout@v4 + - name: Set up Python ${{ matrix.python-version }} + uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python-version }} + - name: Install package and test dependencies + run: | + python -m pip install --upgrade pip + pip install -e . + pip install pytest + - name: Run tests + run: pytest -q + - name: Smoke-test entry points + run: | + domain-analyzer --version + domain-analyzer --help + python -m domain_security_analyzer --version + + build: + name: build & metadata check + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: '3.x' + - name: Build distributions and validate metadata + run: | + python -m pip install --upgrade pip build twine + python -m build + twine check dist/* diff --git a/.gitignore b/.gitignore index 7a60b85..f818088 100644 --- a/.gitignore +++ b/.gitignore @@ -1,2 +1,8 @@ __pycache__/ *.pyc + +# Packaging / build artifacts +build/ +dist/ +*.egg-info/ +*.egg diff --git a/AUTHORS.md b/AUTHORS.md new file mode 100644 index 0000000..ce6b6ba --- /dev/null +++ b/AUTHORS.md @@ -0,0 +1,9 @@ +# Authors + +Domain Security Analyzer is created and maintained by: + +- CallMarcus () + +Thanks also to everyone who has contributed bug reports, ideas, and code. See the +[contributors graph](https://github.com/CallMarcus/domain-security-analyzer/graphs/contributors) +for the full list. diff --git a/CHANGELOG.md b/CHANGELOG.md index ea2ac24..d9d0db5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,12 +1,44 @@ # Changelog -All notable changes to this project will be documented here. +All notable changes to this project are documented here. -## Unreleased +The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), +and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). -- Subdomain discovery: filter out hosts that resolve solely due to wildcard DNS by comparing against A and CNAME baselines. -- Add CLI flag `--include-wildcard-matches` to include wildcard-matched subdomains when desired. -- Add CLI flag `--filtered-subdomains-file ` to export filtered subdomains per domain to a separate CSV. -- Update README and docs with a Wildcard Filtering note and flag usage. -- Add `scripts/test_wildcard_filtering.py` mock-based test harness demonstrating the filtering behavior. +## [Unreleased] +## [1.0.0] - 2026-06-21 + +First packaged release, distributable via PyPI. + +### Added + +- **Packaging**: project is now an installable Python package + (`pip install domain-security-analyzer`) with a `pyproject.toml`, a + `domain-analyzer` console entry point, `python -m domain_security_analyzer` + module execution, and an importable API + (`from domain_security_analyzer import DomainAnalyzer`). +- **Subresource Integrity (SRI) scanning**: detects external JS/CSS resources, + computes SRI coverage, identifies SHA-256/384/512 usage, and reports gaps in + the CSV output. Standalone `scripts/sri_parser.py` crawler mirrors + SecurityScorecard's "Unsafe SRI" guidance. +- **Wildcard subdomain filtering**: suppresses subdomains that resolve solely due + to wildcard DNS by comparing against A and CNAME baselines. +- CLI flag `--include-wildcard-matches` to include wildcard-matched subdomains. +- CLI flag `--filtered-subdomains-file ` to export filtered subdomains per + domain to a separate CSV. +- `scripts/test_wildcard_filtering.py` mock-based test harness demonstrating the + filtering behavior. +- Reference documentation for SRI, CSV output, SPF, DKIM, and DMARC under + `docs/`. + +### Changed + +- Core logic moved from the top-level `domain_analyzer.py` script into the + `domain_security_analyzer` package. `domain_analyzer.py` is retained as a thin + backward-compatible shim, so `python domain_analyzer.py ...` and + `from domain_analyzer import DomainAnalyzer` continue to work. +- README and docs updated with a Wildcard Filtering note and flag usage. + +[Unreleased]: https://github.com/CallMarcus/domain-security-analyzer/compare/v1.0.0...HEAD +[1.0.0]: https://github.com/CallMarcus/domain-security-analyzer/releases/tag/v1.0.0 diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md new file mode 100644 index 0000000..97916d2 --- /dev/null +++ b/CONTRIBUTING.md @@ -0,0 +1,55 @@ +# Contributing + +Thanks for your interest in improving Domain Security Analyzer! Contributions of +all kinds are welcome — bug reports, feature ideas, documentation, and code. + +## Getting started + +1. Fork the repository and clone your fork. +2. Create a development environment and install the package in editable mode: + + ```bash + python -m venv .venv + source .venv/bin/activate # Windows: .venv\Scripts\activate + pip install -e . + ``` + +3. Create a branch for your change: + + ```bash + git checkout -b my-feature + ``` + +## Project layout + +- `domain_security_analyzer/` — the installable package + - `analyzer.py` — core analysis logic (`DomainAnalyzer`, `analyze_domains_from_file`) + - `cli.py` — command-line interface (`domain-analyzer` entry point) +- `domain_analyzer.py` — thin backward-compatible shim for the legacy script path +- `scripts/` — standalone helpers (`sri_parser.py`, `parked_domain_csv.py`, test harnesses) +- `docs/` — reference guides (SRI, CSV output, SPF, DKIM, DMARC) + +## Making changes + +- Keep changes focused and match the style of the surrounding code. +- Update `README.md` and the relevant files in `docs/` when behavior changes. +- Add an entry under the `## [Unreleased]` section of `CHANGELOG.md`. +- Verify the package still builds and the CLI works: + + ```bash + pip install -e . + domain-analyzer --help + python -m build && twine check dist/* + ``` + +## Submitting + +1. Push your branch and open a pull request against `main`. +2. Describe what the change does and why, and link any related issues. +3. Make sure the description notes any new dependencies or CLI flags. + +## Reporting issues + +Open an issue at + with steps to +reproduce, the command you ran, and the observed vs. expected behavior. diff --git a/MANIFEST.in b/MANIFEST.in new file mode 100644 index 0000000..356ccf7 --- /dev/null +++ b/MANIFEST.in @@ -0,0 +1,9 @@ +include README.md +include LICENSE +include CHANGELOG.md +include CONTRIBUTING.md +include AUTHORS.md +include requirements.txt +recursive-include docs *.md +recursive-include examples * +include scripts/*.py diff --git a/README.md b/README.md index becf3c6..a9ca9fa 100644 --- a/README.md +++ b/README.md @@ -1,5 +1,10 @@ # Domain Security Analyzer +[![PyPI version](https://img.shields.io/pypi/v/domain-security-analyzer.svg)](https://pypi.org/project/domain-security-analyzer/) +[![Python versions](https://img.shields.io/pypi/pyversions/domain-security-analyzer.svg)](https://pypi.org/project/domain-security-analyzer/) +[![Downloads](https://img.shields.io/pypi/dm/domain-security-analyzer.svg)](https://pypi.org/project/domain-security-analyzer/) +[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](LICENSE) + A comprehensive Python tool for analyzing domain security configurations including DNS records, email security policies, subdomain discovery, and **Subresource Integrity (SRI) scanning**. The tool performs parallel analysis of domain portfolios to identify potential security configuration issues and modern security compliance gaps. ## Features @@ -39,6 +44,25 @@ A comprehensive Python tool for analyzing domain security configurations includi ### **Installation** +Install from PyPI (recommended): + +```bash +pip install domain-security-analyzer +``` + +This installs the dependencies automatically and adds the `domain-analyzer` +command to your `PATH`. + +#### From source + +```bash +git clone https://github.com/CallMarcus/domain-security-analyzer.git +cd domain-security-analyzer +pip install -e . +``` + +#### Dependencies only (running the script directly) + ```bash # Install required dependencies pip install dnspython requests beautifulsoup4 @@ -47,7 +71,7 @@ pip install dnspython requests beautifulsoup4 pip install -r requirements.txt ``` -The script automatically validates dependencies and provides installation guidance: +When run as a script, the tool automatically validates dependencies and provides installation guidance: ```bash $ python domain_analyzer.py @@ -72,16 +96,25 @@ contoso.com rzy.domain.com ``` -Run the analyzer and specify the output CSV file: +Run the analyzer and specify the output CSV file. If installed from PyPI, use the +`domain-analyzer` command: ```bash -python domain_analyzer.py examples/domains.txt report.csv +domain-analyzer examples/domains.txt report.csv ``` You can optionally set the number of parallel workers: ```bash -python domain_analyzer.py examples/domains.txt report.csv 20 +domain-analyzer examples/domains.txt report.csv 20 +``` + +The same interface is available via `python -m domain_security_analyzer` or, for +backward compatibility, by running the script directly: + +```bash +python -m domain_security_analyzer examples/domains.txt report.csv +python domain_analyzer.py examples/domains.txt report.csv # legacy shim ``` The generated CSV includes comprehensive security analysis with **29 columns**: diff --git a/domain_analyzer.py b/domain_analyzer.py index cb944aa..cac55dd 100644 --- a/domain_analyzer.py +++ b/domain_analyzer.py @@ -1,663 +1,24 @@ -import sys -from datetime import datetime +#!/usr/bin/env python3 +"""Backward-compatible entry point. -# Check for required modules before proceeding +The implementation now lives in the :mod:`domain_security_analyzer` package. +This shim preserves the historical ``python domain_analyzer.py ...`` invocation +and the ``from domain_analyzer import DomainAnalyzer`` import path. +Prefer the installed console script ``domain-analyzer`` or +``python -m domain_security_analyzer`` going forward. +""" -def check_required_modules(): - """Check if required modules are installed and provide installation instructions if missing.""" - missing_modules = [] - - try: - import dns.resolver - except ImportError: - missing_modules.append('dnspython') - - try: - import requests - except ImportError: - missing_modules.append('requests') - - try: - from bs4 import BeautifulSoup - except ImportError: - missing_modules.append('beautifulsoup4') - - if missing_modules: - print("ERROR: Missing required Python packages!") - print("\nPlease install the following packages:") - for module in missing_modules: - print(f" - {module}") - - print("\nInstallation command:") - print(f" pip install {' '.join(missing_modules)}") - print("\nOr if using pip3:") - print(f" pip3 install {' '.join(missing_modules)}") - print("\nIf using a virtual environment, activate it first and then run the pip command.") - - if 'beautifulsoup4' in missing_modules: - print("\nNote: beautifulsoup4 is required for SRI (Subresource Integrity) analysis") - sys.exit(1) +from domain_security_analyzer.cli import main -# Check modules before importing -check_required_modules() -import dns.resolver -import requests -import concurrent.futures -import csv -from typing import Dict, List, Optional -from bs4 import BeautifulSoup -from urllib.parse import urlparse +def __getattr__(name): + # Lazily re-export the public API (DomainAnalyzer, analyze_domains_from_file) + # so importing this shim does not require the runtime dependencies unless the + # analysis code is actually used. + from domain_security_analyzer import analyzer + return getattr(analyzer, name) -class DomainAnalyzer: - def __init__(self, include_wildcard_matches: bool = False, collect_filtered: bool = False): - self.resolver = dns.resolver.Resolver() - self.resolver.timeout = 5 - self.resolver.lifetime = 5 - self.include_wildcard_matches = include_wildcard_matches - self.collect_filtered = collect_filtered - - # Common subdomain prefixes to check - self.common_subdomains = [ - 'www', 'mail', 'webmail', 'email', 'remote', 'portal', 'owa', - 'vpn', 'mta', 'mx', 'imap', 'smtp', 'pop', 'cp', 'cpanel', - 'webdisk', 'whm', 'ns1', 'ns2', 'autodiscover', 'autoconfig', - 'admin', 'cloud', 'dev', 'ftp', 'test', 'staging' - ] - - # Common hosting providers' default records - self.hosting_patterns = { - 'GoDaddy': ['.secureserver.net'], - 'BlueHost': ['.bluehost.com'], - 'HostGator': ['.hostgator.com'], - 'DreamHost': ['.dreamhost.com'], - 'NameCheap': ['.registrar-servers.com'], - 'OVH': ['.ovh.net'], - 'AWS': ['.amazonaws.com'], - 'Google Cloud': ['.googlehosted.com'], - 'Microsoft Azure': ['.azurewebsites.net'], - 'Cloudflare': ['.cloudflare.net'] - } - - def get_dns_record(self, domain: str, record_type: str) -> Optional[List[str]]: - """Query DNS records of specified type for a domain.""" - try: - # Try with default resolver first - answers = self.resolver.resolve(domain, record_type) - return [str(rdata) for rdata in answers] - except (dns.resolver.NXDOMAIN, dns.resolver.NoAnswer): - return None - except dns.exception.Timeout: - # On timeout, try with system DNS servers - try: - # Get system DNS servers (useful especially on Windows) - system_resolver = dns.resolver.Resolver(configure=True) - system_resolver.timeout = 3 - system_resolver.lifetime = 3 - answers = system_resolver.resolve(domain, record_type) - return [str(rdata) for rdata in answers] - except: - return None - except Exception as e: - if "SERVFAIL" in str(e): - return None # Common on Windows when DNS server is unreachable - return f"Error: {str(e)}" - - def check_spf(self, domain: str) -> Dict: - """Check SPF record for domain.""" - records = self.get_dns_record(domain, 'TXT') - if not records: - return {"exists": False, "record": None} - - spf_records = [r for r in records if r.startswith('"v=spf1')] - if not spf_records: - return {"exists": False, "record": None} - - return { - "exists": True, - "record": spf_records[0], - "multiple_records": len(spf_records) > 1 - } - - def check_dkim(self, domain: str, selectors: List[str] = ['default', 'google', 'dkim', 'k1']) -> Dict: - """Check DKIM record for domain with multiple common selectors.""" - results = [] - for selector in selectors: - dkim_domain = f"{selector}._domainkey.{domain}" - record = self.get_dns_record(dkim_domain, 'TXT') - if record: - results.append({ - "selector": selector, - "record": record[0] - }) - - return { - "exists": bool(results), - "records": results - } - - def check_dmarc(self, domain: str) -> Dict: - """Check DMARC record for domain.""" - dmarc_domain = f"_dmarc.{domain}" - record = self.get_dns_record(dmarc_domain, 'TXT') - - return { - "exists": bool(record), - "record": record[0] if record else None - } - - def discover_subdomains(self, domain: str) -> Dict: - """Discover subdomains using various methods.""" - found_subdomains = set() - filtered_subdomains = set() - cname_records = {} - - # Helper to normalize DNS rrsets for comparison - def _norm_rrset(rrset: Optional[List[str]]) -> Optional[tuple]: - if not rrset: - return None - try: - return tuple(sorted(str(r).strip().lower() for r in rrset)) - except Exception: - return tuple(sorted(rrset)) - - # Establish wildcard baseline answers (if any) - try: - random_sub = f"wildcard-test-{datetime.now().strftime('%Y%m%d%H%M%S')}.{domain}" - wildcard_a = self.get_dns_record(random_sub, 'A') - wildcard_cname = self.get_dns_record(random_sub, 'CNAME') - has_wildcard = bool(wildcard_a or wildcard_cname) - wildcard_a_norm = _norm_rrset(wildcard_a) - wildcard_cname_norm = _norm_rrset(wildcard_cname) - except Exception: - has_wildcard = False - wildcard_a_norm = None - wildcard_cname_norm = None - - # Check common subdomains - for subdomain in self.common_subdomains: - fqdn = f"{subdomain}.{domain}" - try: - # Prefer explicit CNAMEs - cname = self.get_dns_record(fqdn, 'CNAME') - a_records = self.get_dns_record(fqdn, 'A') - - include = False - - if cname: - # Include CNAMEs unless they match the wildcard CNAME baseline - if has_wildcard and _norm_rrset(cname) == wildcard_cname_norm and not self.include_wildcard_matches: - include = False - if self.collect_filtered: - filtered_subdomains.add(fqdn) - else: - include = True - cname_records[fqdn] = cname[0] - elif a_records: - # If wildcard is present, suppress A-only results that - # exactly match the wildcard baseline to avoid false positives - if has_wildcard: - if self.include_wildcard_matches or _norm_rrset(a_records) != wildcard_a_norm: - include = True - else: - if self.collect_filtered: - filtered_subdomains.add(fqdn) - else: - include = True - - if include: - found_subdomains.add(fqdn) - except Exception: - continue - - # Identify hosting provider - hosting_provider = None - for provider, patterns in self.hosting_patterns.items(): - for pattern in patterns: - if any(pattern in cname for cname in cname_records.values()): - hosting_provider = provider - break - if hosting_provider: - break - - return { - "subdomains": list(found_subdomains), - "cname_records": cname_records, - "has_wildcard_dns": has_wildcard, - "hosting_provider": hosting_provider, - "filtered_subdomains": list(filtered_subdomains) - } - - def check_http_redirect(self, domain: str) -> tuple[Dict, str]: - """Check for insecure HTTP to HTTPS redirects and capture HTML content.""" - result = { - "http_accessible": False, - "redirects_to_https": False, - "final_url": None, - "error": None, - "redirect_chain": [] - } - html_content = "" - - try: - http_url = f"http://{domain}" - response = requests.get(http_url, allow_redirects=True, timeout=10) - - result["http_accessible"] = True - result["final_url"] = response.url - result["redirects_to_https"] = response.url.startswith("https://") - - # Capture redirect chain - if response.history: - result["redirect_chain"] = [r.url for r in response.history] - result["redirect_chain"].append(response.url) - - # Capture HTML content for SRI analysis (limit to reasonable size) - if response.headers.get('content-type', '').startswith('text/html'): - html_content = response.text[:500000] # Limit to 500KB to avoid memory issues - - except requests.exceptions.RequestException as e: - result["error"] = str(e) - - return result, html_content - - def _is_external_resource(self, url: str, domain: str) -> bool: - """Check if a resource URL is external to the given domain.""" - if not url: - return False - - # Handle relative URLs - if not url.startswith(('http://', 'https://')): - return False - - parsed_url = urlparse(url) - resource_domain = parsed_url.netloc.lower() - - # Remove www prefix for comparison - main_domain = domain.lower().replace('www.', '') - resource_domain = resource_domain.replace('www.', '') - - return resource_domain != main_domain - - def _extract_hash_algorithm(self, integrity_attr: str) -> str: - """Extract hash algorithm from integrity attribute.""" - if not integrity_attr: - return None - - # integrity="sha384-..." or "sha256-..." etc. - if integrity_attr.startswith('sha256-'): - return 'sha256' - elif integrity_attr.startswith('sha384-'): - return 'sha384' - elif integrity_attr.startswith('sha512-'): - return 'sha512' - else: - return 'unknown' - - def check_sri(self, domain: str, html_content: str) -> Dict: - """Analyze Subresource Integrity implementation from HTML content.""" - result = { - "sri_enabled": False, - "total_external_resources": 0, - "resources_with_sri": 0, - "sri_coverage_percentage": 0, - "missing_sri_count": 0, - "sri_algorithms_used": set(), - "error": None - } - - if not html_content: - result["error"] = "No HTML content available" - return result - - try: - soup = BeautifulSoup(html_content, 'html.parser') - external_resources = [] - - # Find external scripts - for script in soup.find_all('script', src=True): - src = script.get('src') - if self._is_external_resource(src, domain): - external_resources.append({ - 'type': 'script', - 'src': src, - 'integrity': script.get('integrity'), - 'crossorigin': script.get('crossorigin') - }) - - # Find external stylesheets - for link in soup.find_all('link', href=True): - if link.get('rel') == ['stylesheet'] or 'stylesheet' in (link.get('rel') or []): - href = link.get('href') - if self._is_external_resource(href, domain): - external_resources.append({ - 'type': 'stylesheet', - 'src': href, - 'integrity': link.get('integrity'), - 'crossorigin': link.get('crossorigin') - }) - - # Analyze SRI implementation - result["total_external_resources"] = len(external_resources) - - for resource in external_resources: - if resource['integrity']: - result["resources_with_sri"] += 1 - algorithm = self._extract_hash_algorithm(resource['integrity']) - if algorithm: - result["sri_algorithms_used"].add(algorithm) - - result["missing_sri_count"] = result["total_external_resources"] - result["resources_with_sri"] - - if result["total_external_resources"] > 0: - result["sri_coverage_percentage"] = round( - (result["resources_with_sri"] / result["total_external_resources"]) * 100, 1 - ) - result["sri_enabled"] = result["resources_with_sri"] > 0 - - # Convert set to sorted list for CSV output - result["sri_algorithms_used"] = sorted(list(result["sri_algorithms_used"])) - - except Exception as e: - result["error"] = f"SRI parsing error: {str(e)}" - - return result - - def get_parent_domain(self, domain: str) -> str: - """Extract parent domain from subdomain (e.g., www.example.com -> example.com).""" - parts = domain.split('.') - if len(parts) <= 2: - return domain # Already a parent domain - - # Handle common TLDs and country codes - # For simplicity, assume last two parts are the parent domain - # This works for most cases like .com, .org, .co.uk, etc. - return '.'.join(parts[-2:]) - - def get_soa_record(self, domain: str) -> Dict: - """Get SOA (Start of Authority) record for the parent domain.""" - parent_domain = self.get_parent_domain(domain) - - try: - soa_records = self.get_dns_record(parent_domain, 'SOA') - if not soa_records: - return {"exists": False, "parent_domain": parent_domain, "record": None} - - # Parse SOA record components - only extract DNS names - soa_parts = soa_records[0].split() - if len(soa_parts) >= 2: - # Only include the primary nameserver and admin email (first two fields) - dns_names_only = f"{soa_parts[0]} {soa_parts[1]}" - return { - "exists": True, - "parent_domain": parent_domain, - "record": dns_names_only, - "primary_ns": soa_parts[0], - "admin_email": soa_parts[1] - } - else: - return { - "exists": True, - "parent_domain": parent_domain, - "record": soa_records[0], - "primary_ns": None, - "admin_email": None - } - except Exception as e: - return { - "exists": False, - "parent_domain": parent_domain, - "record": None, - "error": str(e) - } - - def analyze_domain(self, domain: str) -> Dict: - """Perform complete analysis of a domain.""" - subdomain_info = self.discover_subdomains(domain) - - # Get HTTP redirect info and HTML content in one request - http_redirect_info, html_content = self.check_http_redirect(domain) - - # Analyze SRI using the captured HTML content - sri_info = self.check_sri(domain, html_content) - - return { - "domain": domain, - "timestamp": datetime.now().isoformat(), - "soa": self.get_soa_record(domain), - "spf": self.check_spf(domain), - "dkim": self.check_dkim(domain), - "dmarc": self.check_dmarc(domain), - "subdomains": subdomain_info, - "http_redirect": http_redirect_info, - "sri": sri_info - } - -def analyze_domains_from_file(input_file: str, output_file: str, max_workers: int = 10, *, include_wildcard_matches: bool = False, filtered_subdomains_file: Optional[str] = None): - """Analyze multiple domains from a file and save results to CSV.""" - - # Read domains from input file - with open(input_file, 'r') as f: - domains = [line.strip() for line in f if line.strip()] - - total_domains = len(domains) - completed = 0 - - def analyze_single_domain(domain: str) -> Dict: - """Worker function for parallel processing""" - nonlocal completed - analyzer = DomainAnalyzer(include_wildcard_matches=include_wildcard_matches, collect_filtered=bool(filtered_subdomains_file)) # Create new instance for thread safety - try: - result = analyzer.analyze_domain(domain) - except Exception as e: - # Create error result with all required fields for CSV - result = { - "domain": domain, - "timestamp": datetime.now().isoformat(), - "error": str(e), - "soa": {"exists": False, "parent_domain": domain, "record": None, "primary_ns": None, "admin_email": None}, - "spf": {"exists": False, "record": None}, - "dkim": {"exists": False, "records": []}, - "dmarc": {"exists": False, "record": None}, - "subdomains": {"subdomains": [], "cname_records": {}, "has_wildcard_dns": False, "hosting_provider": None, "filtered_subdomains": []}, - "http_redirect": {"http_accessible": False, "redirects_to_https": False, "final_url": None, "error": str(e), "redirect_chain": []}, - "sri": {"sri_enabled": False, "total_external_resources": 0, "resources_with_sri": 0, "sri_coverage_percentage": 0, "missing_sri_count": 0, "sri_algorithms_used": [], "error": "Domain analysis failed"} - } - - completed += 1 - print(f"Progress: {completed}/{total_domains} domains analyzed ({(completed/total_domains)*100:.1f}%)") - return result - - results = [] - print(f"Starting analysis of {total_domains} domains using {max_workers} parallel workers...") - - # Use ThreadPoolExecutor for parallel processing - with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor: - future_to_domain = {executor.submit(analyze_single_domain, domain): domain for domain in domains} - - for future in concurrent.futures.as_completed(future_to_domain): - domain = future_to_domain[future] - try: - result = future.result() - results.append(result) - except Exception as e: - print(f"Error analyzing {domain}: {str(e)}") - # Create error result with all required fields for CSV - error_result = { - "domain": domain, - "timestamp": datetime.now().isoformat(), - "error": str(e), - "soa": {"exists": False, "parent_domain": domain, "record": None, "primary_ns": None, "admin_email": None}, - "spf": {"exists": False, "record": None}, - "dkim": {"exists": False, "records": []}, - "dmarc": {"exists": False, "record": None}, - "subdomains": {"subdomains": [], "cname_records": {}, "has_wildcard_dns": False, "hosting_provider": None}, - "http_redirect": {"http_accessible": False, "redirects_to_https": False, "final_url": None, "error": str(e), "redirect_chain": []}, - "sri": {"sri_enabled": False, "total_external_resources": 0, "resources_with_sri": 0, "sri_coverage_percentage": 0, "missing_sri_count": 0, "sri_algorithms_used": [], "error": "Domain analysis failed"} - } - results.append(error_result) - - # Write results to CSV - with open(output_file, 'w', newline='') as f: - writer = csv.writer(f) - # Write header - writer.writerow([ - 'Domain', - 'Timestamp', - 'Parent Domain', - 'SOA Exists', - 'SOA Record', - 'Primary NS', - 'Admin Email', - 'SPF Exists', - 'SPF Record', - 'DKIM Exists', - 'DKIM Records', - 'DMARC Exists', - 'DMARC Record', - 'Discovered Subdomains', - 'CNAME Records', - 'Has Wildcard DNS', - 'Hosting Provider', - 'HTTP Accessible', - 'Redirects to HTTPS', - 'Final URL', - 'Redirect Chain', - 'HTTP Error', - 'SRI Enabled', - 'Total External Resources', - 'Resources With SRI', - 'SRI Coverage %', - 'Missing SRI Count', - 'SRI Algorithms Used', - 'SRI Error' - ]) - - # Write results - for r in results: - writer.writerow([ - r['domain'], - r['timestamp'], - r['soa']['parent_domain'], - r['soa']['exists'], - r['soa'].get('record'), - r['soa'].get('primary_ns'), - r['soa'].get('admin_email'), - r['spf']['exists'], - r['spf'].get('record'), - r['dkim']['exists'], - ';'.join([f"{rec['selector']}:{rec['record']}" for rec in r['dkim']['records']]) if r['dkim']['records'] else '', - r['dmarc']['exists'], - r['dmarc'].get('record'), - ','.join(r['subdomains']['subdomains']), - ','.join([f"{k}:{v}" for k,v in r['subdomains']['cname_records'].items()]), - r['subdomains']['has_wildcard_dns'], - r['subdomains']['hosting_provider'], - r['http_redirect']['http_accessible'], - r['http_redirect']['redirects_to_https'], - r['http_redirect']['final_url'], - ' -> '.join(r['http_redirect'].get('redirect_chain', [])), - r['http_redirect']['error'], - r['sri']['sri_enabled'], - r['sri']['total_external_resources'], - r['sri']['resources_with_sri'], - r['sri']['sri_coverage_percentage'], - r['sri']['missing_sri_count'], - ','.join(r['sri']['sri_algorithms_used']) if r['sri']['sri_algorithms_used'] else '', - r['sri']['error'] - ]) - - # Optionally write filtered subdomains to a separate CSV - if filtered_subdomains_file: - with open(filtered_subdomains_file, 'w', newline='') as f: - writer = csv.writer(f) - writer.writerow(['Domain', 'Filtered Subdomains']) - for r in results: - filtered = r.get('subdomains', {}).get('filtered_subdomains', []) - if filtered: - writer.writerow([r['domain'], ','.join(filtered)]) if __name__ == "__main__": - import platform - import os - - # Windows-specific console configuration - if platform.system() == 'Windows': - try: - import colorama - colorama.init() # Initialize colorama for Windows color support - except ImportError: - pass # colorama not installed, colors won't work - - # Try to set console to UTF-8 mode - try: - os.system('chcp 65001 > nul') - except: - pass - - if len(sys.argv) < 3: - print("Usage: python domain_analyzer.py input_file.txt output_file.csv [max_workers] [--include-wildcard-matches] [--filtered-subdomains-file path]") - print("max_workers: Optional, default is OS-dependent") - sys.exit(1) - - input_file = sys.argv[1] - output_file = sys.argv[2] - - # Adjust default workers based on OS and CPU count - default_workers = min(10, (os.cpu_count() or 4) * 2) - max_workers = None - include_wildcard_matches = False - filtered_subdomains_file = None - - # Parse optional args - remaining = sys.argv[3:] - # First, try to parse a positional max_workers if it's an int - if remaining: - try: - max_workers = int(remaining[0]) - remaining = remaining[1:] - except ValueError: - max_workers = None - - max_workers = max_workers or default_workers - - i = 0 - while i < len(remaining): - arg = remaining[i] - if arg == '--include-wildcard-matches': - include_wildcard_matches = True - i += 1 - elif arg == '--filtered-subdomains-file': - if i + 1 >= len(remaining): - print("Error: --filtered-subdomains-file requires a path argument") - sys.exit(1) - filtered_subdomains_file = os.path.normpath(remaining[i+1]) - i += 2 - else: - print(f"Unrecognized argument: {arg}") - print("Usage: python domain_analyzer.py input_file.txt output_file.csv [max_workers] [--include-wildcard-matches] [--filtered-subdomains-file path]") - sys.exit(1) - - # Ensure input/output files use proper path separators - input_file = os.path.normpath(input_file) - output_file = os.path.normpath(output_file) - - print("\nStarting domain analysis:") - print(f"Operating System: {platform.system()} {platform.release()}") - print(f"Input file: {input_file}") - print(f"Output file: {output_file}") - print(f"Workers: {max_workers}") - if include_wildcard_matches: - print("Include wildcard-matched subdomains: True") - if filtered_subdomains_file: - print(f"Filtered subdomains file: {filtered_subdomains_file}") - print("") - - try: - analyze_domains_from_file(input_file, output_file, max_workers, include_wildcard_matches=include_wildcard_matches, filtered_subdomains_file=filtered_subdomains_file) - except KeyboardInterrupt: - print("\nAnalysis interrupted by user. Partial results may have been saved.") - except Exception as e: - print(f"\nError during analysis: {str(e)}") - sys.exit(1) + main() diff --git a/domain_security_analyzer/__init__.py b/domain_security_analyzer/__init__.py new file mode 100644 index 0000000..fda8d34 --- /dev/null +++ b/domain_security_analyzer/__init__.py @@ -0,0 +1,11 @@ +"""Domain Security Analyzer. + +A tool for analyzing domain security configurations including DNS records, +email authentication (SPF/DKIM/DMARC), subdomain discovery, and Subresource +Integrity (SRI) scanning. +""" + +from .__version__ import __version__ +from .analyzer import DomainAnalyzer, analyze_domains_from_file + +__all__ = ["DomainAnalyzer", "analyze_domains_from_file", "__version__"] diff --git a/domain_security_analyzer/__main__.py b/domain_security_analyzer/__main__.py new file mode 100644 index 0000000..2d7d705 --- /dev/null +++ b/domain_security_analyzer/__main__.py @@ -0,0 +1,6 @@ +"""Enable ``python -m domain_security_analyzer``.""" + +from .cli import main + +if __name__ == "__main__": + main() diff --git a/domain_security_analyzer/__version__.py b/domain_security_analyzer/__version__.py new file mode 100644 index 0000000..5becc17 --- /dev/null +++ b/domain_security_analyzer/__version__.py @@ -0,0 +1 @@ +__version__ = "1.0.0" diff --git a/domain_security_analyzer/analyzer.py b/domain_security_analyzer/analyzer.py new file mode 100644 index 0000000..29a7997 --- /dev/null +++ b/domain_security_analyzer/analyzer.py @@ -0,0 +1,546 @@ +"""Core domain analysis logic. + +Contains :class:`DomainAnalyzer` and :func:`analyze_domains_from_file`, the +importable API for the package. The command-line interface lives in +``domain_security_analyzer.cli``. +""" + +import concurrent.futures +import csv +from datetime import datetime +from typing import Dict, List, Optional +from urllib.parse import urlparse + +import dns.resolver +import requests +from bs4 import BeautifulSoup + + +class DomainAnalyzer: + def __init__(self, include_wildcard_matches: bool = False, collect_filtered: bool = False): + self.resolver = dns.resolver.Resolver() + self.resolver.timeout = 5 + self.resolver.lifetime = 5 + self.include_wildcard_matches = include_wildcard_matches + self.collect_filtered = collect_filtered + + # Common subdomain prefixes to check + self.common_subdomains = [ + 'www', 'mail', 'webmail', 'email', 'remote', 'portal', 'owa', + 'vpn', 'mta', 'mx', 'imap', 'smtp', 'pop', 'cp', 'cpanel', + 'webdisk', 'whm', 'ns1', 'ns2', 'autodiscover', 'autoconfig', + 'admin', 'cloud', 'dev', 'ftp', 'test', 'staging' + ] + + # Common hosting providers' default records + self.hosting_patterns = { + 'GoDaddy': ['.secureserver.net'], + 'BlueHost': ['.bluehost.com'], + 'HostGator': ['.hostgator.com'], + 'DreamHost': ['.dreamhost.com'], + 'NameCheap': ['.registrar-servers.com'], + 'OVH': ['.ovh.net'], + 'AWS': ['.amazonaws.com'], + 'Google Cloud': ['.googlehosted.com'], + 'Microsoft Azure': ['.azurewebsites.net'], + 'Cloudflare': ['.cloudflare.net'] + } + + def get_dns_record(self, domain: str, record_type: str) -> Optional[List[str]]: + """Query DNS records of specified type for a domain.""" + try: + # Try with default resolver first + answers = self.resolver.resolve(domain, record_type) + return [str(rdata) for rdata in answers] + except (dns.resolver.NXDOMAIN, dns.resolver.NoAnswer): + return None + except dns.exception.Timeout: + # On timeout, try with system DNS servers + try: + # Get system DNS servers (useful especially on Windows) + system_resolver = dns.resolver.Resolver(configure=True) + system_resolver.timeout = 3 + system_resolver.lifetime = 3 + answers = system_resolver.resolve(domain, record_type) + return [str(rdata) for rdata in answers] + except: + return None + except Exception as e: + if "SERVFAIL" in str(e): + return None # Common on Windows when DNS server is unreachable + return f"Error: {str(e)}" + + def check_spf(self, domain: str) -> Dict: + """Check SPF record for domain.""" + records = self.get_dns_record(domain, 'TXT') + if not records: + return {"exists": False, "record": None} + + spf_records = [r for r in records if r.startswith('"v=spf1')] + if not spf_records: + return {"exists": False, "record": None} + + return { + "exists": True, + "record": spf_records[0], + "multiple_records": len(spf_records) > 1 + } + + def check_dkim(self, domain: str, selectors: List[str] = ['default', 'google', 'dkim', 'k1']) -> Dict: + """Check DKIM record for domain with multiple common selectors.""" + results = [] + for selector in selectors: + dkim_domain = f"{selector}._domainkey.{domain}" + record = self.get_dns_record(dkim_domain, 'TXT') + if record: + results.append({ + "selector": selector, + "record": record[0] + }) + + return { + "exists": bool(results), + "records": results + } + + def check_dmarc(self, domain: str) -> Dict: + """Check DMARC record for domain.""" + dmarc_domain = f"_dmarc.{domain}" + record = self.get_dns_record(dmarc_domain, 'TXT') + + return { + "exists": bool(record), + "record": record[0] if record else None + } + + def discover_subdomains(self, domain: str) -> Dict: + """Discover subdomains using various methods.""" + found_subdomains = set() + filtered_subdomains = set() + cname_records = {} + + # Helper to normalize DNS rrsets for comparison + def _norm_rrset(rrset: Optional[List[str]]) -> Optional[tuple]: + if not rrset: + return None + try: + return tuple(sorted(str(r).strip().lower() for r in rrset)) + except Exception: + return tuple(sorted(rrset)) + + # Establish wildcard baseline answers (if any) + try: + random_sub = f"wildcard-test-{datetime.now().strftime('%Y%m%d%H%M%S')}.{domain}" + wildcard_a = self.get_dns_record(random_sub, 'A') + wildcard_cname = self.get_dns_record(random_sub, 'CNAME') + has_wildcard = bool(wildcard_a or wildcard_cname) + wildcard_a_norm = _norm_rrset(wildcard_a) + wildcard_cname_norm = _norm_rrset(wildcard_cname) + except Exception: + has_wildcard = False + wildcard_a_norm = None + wildcard_cname_norm = None + + # Check common subdomains + for subdomain in self.common_subdomains: + fqdn = f"{subdomain}.{domain}" + try: + # Prefer explicit CNAMEs + cname = self.get_dns_record(fqdn, 'CNAME') + a_records = self.get_dns_record(fqdn, 'A') + + include = False + + if cname: + # Include CNAMEs unless they match the wildcard CNAME baseline + if has_wildcard and _norm_rrset(cname) == wildcard_cname_norm and not self.include_wildcard_matches: + include = False + if self.collect_filtered: + filtered_subdomains.add(fqdn) + else: + include = True + cname_records[fqdn] = cname[0] + elif a_records: + # If wildcard is present, suppress A-only results that + # exactly match the wildcard baseline to avoid false positives + if has_wildcard: + if self.include_wildcard_matches or _norm_rrset(a_records) != wildcard_a_norm: + include = True + else: + if self.collect_filtered: + filtered_subdomains.add(fqdn) + else: + include = True + + if include: + found_subdomains.add(fqdn) + except Exception: + continue + + # Identify hosting provider + hosting_provider = None + for provider, patterns in self.hosting_patterns.items(): + for pattern in patterns: + if any(pattern in cname for cname in cname_records.values()): + hosting_provider = provider + break + if hosting_provider: + break + + return { + "subdomains": list(found_subdomains), + "cname_records": cname_records, + "has_wildcard_dns": has_wildcard, + "hosting_provider": hosting_provider, + "filtered_subdomains": list(filtered_subdomains) + } + + def check_http_redirect(self, domain: str) -> "tuple[Dict, str]": + """Check for insecure HTTP to HTTPS redirects and capture HTML content.""" + result = { + "http_accessible": False, + "redirects_to_https": False, + "final_url": None, + "error": None, + "redirect_chain": [] + } + html_content = "" + + try: + http_url = f"http://{domain}" + response = requests.get(http_url, allow_redirects=True, timeout=10) + + result["http_accessible"] = True + result["final_url"] = response.url + result["redirects_to_https"] = response.url.startswith("https://") + + # Capture redirect chain + if response.history: + result["redirect_chain"] = [r.url for r in response.history] + result["redirect_chain"].append(response.url) + + # Capture HTML content for SRI analysis (limit to reasonable size) + if response.headers.get('content-type', '').startswith('text/html'): + html_content = response.text[:500000] # Limit to 500KB to avoid memory issues + + except requests.exceptions.RequestException as e: + result["error"] = str(e) + + return result, html_content + + def _is_external_resource(self, url: str, domain: str) -> bool: + """Check if a resource URL is external to the given domain.""" + if not url: + return False + + # Handle relative URLs + if not url.startswith(('http://', 'https://')): + return False + + parsed_url = urlparse(url) + resource_domain = parsed_url.netloc.lower() + + # Remove www prefix for comparison + main_domain = domain.lower().replace('www.', '') + resource_domain = resource_domain.replace('www.', '') + + return resource_domain != main_domain + + def _extract_hash_algorithm(self, integrity_attr: str) -> str: + """Extract hash algorithm from integrity attribute.""" + if not integrity_attr: + return None + + # integrity="sha384-..." or "sha256-..." etc. + if integrity_attr.startswith('sha256-'): + return 'sha256' + elif integrity_attr.startswith('sha384-'): + return 'sha384' + elif integrity_attr.startswith('sha512-'): + return 'sha512' + else: + return 'unknown' + + def check_sri(self, domain: str, html_content: str) -> Dict: + """Analyze Subresource Integrity implementation from HTML content.""" + result = { + "sri_enabled": False, + "total_external_resources": 0, + "resources_with_sri": 0, + "sri_coverage_percentage": 0, + "missing_sri_count": 0, + "sri_algorithms_used": set(), + "error": None + } + + if not html_content: + result["error"] = "No HTML content available" + return result + + try: + soup = BeautifulSoup(html_content, 'html.parser') + external_resources = [] + + # Find external scripts + for script in soup.find_all('script', src=True): + src = script.get('src') + if self._is_external_resource(src, domain): + external_resources.append({ + 'type': 'script', + 'src': src, + 'integrity': script.get('integrity'), + 'crossorigin': script.get('crossorigin') + }) + + # Find external stylesheets + for link in soup.find_all('link', href=True): + if link.get('rel') == ['stylesheet'] or 'stylesheet' in (link.get('rel') or []): + href = link.get('href') + if self._is_external_resource(href, domain): + external_resources.append({ + 'type': 'stylesheet', + 'src': href, + 'integrity': link.get('integrity'), + 'crossorigin': link.get('crossorigin') + }) + + # Analyze SRI implementation + result["total_external_resources"] = len(external_resources) + + for resource in external_resources: + if resource['integrity']: + result["resources_with_sri"] += 1 + algorithm = self._extract_hash_algorithm(resource['integrity']) + if algorithm: + result["sri_algorithms_used"].add(algorithm) + + result["missing_sri_count"] = result["total_external_resources"] - result["resources_with_sri"] + + if result["total_external_resources"] > 0: + result["sri_coverage_percentage"] = round( + (result["resources_with_sri"] / result["total_external_resources"]) * 100, 1 + ) + result["sri_enabled"] = result["resources_with_sri"] > 0 + + # Convert set to sorted list for CSV output + result["sri_algorithms_used"] = sorted(list(result["sri_algorithms_used"])) + + except Exception as e: + result["error"] = f"SRI parsing error: {str(e)}" + + return result + + def get_parent_domain(self, domain: str) -> str: + """Extract parent domain from subdomain (e.g., www.example.com -> example.com).""" + parts = domain.split('.') + if len(parts) <= 2: + return domain # Already a parent domain + + # Handle common TLDs and country codes + # For simplicity, assume last two parts are the parent domain + # This works for most cases like .com, .org, .co.uk, etc. + return '.'.join(parts[-2:]) + + def get_soa_record(self, domain: str) -> Dict: + """Get SOA (Start of Authority) record for the parent domain.""" + parent_domain = self.get_parent_domain(domain) + + try: + soa_records = self.get_dns_record(parent_domain, 'SOA') + if not soa_records: + return {"exists": False, "parent_domain": parent_domain, "record": None} + + # Parse SOA record components - only extract DNS names + soa_parts = soa_records[0].split() + if len(soa_parts) >= 2: + # Only include the primary nameserver and admin email (first two fields) + dns_names_only = f"{soa_parts[0]} {soa_parts[1]}" + return { + "exists": True, + "parent_domain": parent_domain, + "record": dns_names_only, + "primary_ns": soa_parts[0], + "admin_email": soa_parts[1] + } + else: + return { + "exists": True, + "parent_domain": parent_domain, + "record": soa_records[0], + "primary_ns": None, + "admin_email": None + } + except Exception as e: + return { + "exists": False, + "parent_domain": parent_domain, + "record": None, + "error": str(e) + } + + def analyze_domain(self, domain: str) -> Dict: + """Perform complete analysis of a domain.""" + subdomain_info = self.discover_subdomains(domain) + + # Get HTTP redirect info and HTML content in one request + http_redirect_info, html_content = self.check_http_redirect(domain) + + # Analyze SRI using the captured HTML content + sri_info = self.check_sri(domain, html_content) + + return { + "domain": domain, + "timestamp": datetime.now().isoformat(), + "soa": self.get_soa_record(domain), + "spf": self.check_spf(domain), + "dkim": self.check_dkim(domain), + "dmarc": self.check_dmarc(domain), + "subdomains": subdomain_info, + "http_redirect": http_redirect_info, + "sri": sri_info + } + + +def analyze_domains_from_file(input_file: str, output_file: str, max_workers: int = 10, *, include_wildcard_matches: bool = False, filtered_subdomains_file: Optional[str] = None): + """Analyze multiple domains from a file and save results to CSV.""" + + # Read domains from input file + with open(input_file, 'r') as f: + domains = [line.strip() for line in f if line.strip()] + + total_domains = len(domains) + completed = 0 + + def analyze_single_domain(domain: str) -> Dict: + """Worker function for parallel processing""" + nonlocal completed + analyzer = DomainAnalyzer(include_wildcard_matches=include_wildcard_matches, collect_filtered=bool(filtered_subdomains_file)) # Create new instance for thread safety + try: + result = analyzer.analyze_domain(domain) + except Exception as e: + # Create error result with all required fields for CSV + result = { + "domain": domain, + "timestamp": datetime.now().isoformat(), + "error": str(e), + "soa": {"exists": False, "parent_domain": domain, "record": None, "primary_ns": None, "admin_email": None}, + "spf": {"exists": False, "record": None}, + "dkim": {"exists": False, "records": []}, + "dmarc": {"exists": False, "record": None}, + "subdomains": {"subdomains": [], "cname_records": {}, "has_wildcard_dns": False, "hosting_provider": None, "filtered_subdomains": []}, + "http_redirect": {"http_accessible": False, "redirects_to_https": False, "final_url": None, "error": str(e), "redirect_chain": []}, + "sri": {"sri_enabled": False, "total_external_resources": 0, "resources_with_sri": 0, "sri_coverage_percentage": 0, "missing_sri_count": 0, "sri_algorithms_used": [], "error": "Domain analysis failed"} + } + + completed += 1 + print(f"Progress: {completed}/{total_domains} domains analyzed ({(completed/total_domains)*100:.1f}%)") + return result + + results = [] + print(f"Starting analysis of {total_domains} domains using {max_workers} parallel workers...") + + # Use ThreadPoolExecutor for parallel processing + with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor: + future_to_domain = {executor.submit(analyze_single_domain, domain): domain for domain in domains} + + for future in concurrent.futures.as_completed(future_to_domain): + domain = future_to_domain[future] + try: + result = future.result() + results.append(result) + except Exception as e: + print(f"Error analyzing {domain}: {str(e)}") + # Create error result with all required fields for CSV + error_result = { + "domain": domain, + "timestamp": datetime.now().isoformat(), + "error": str(e), + "soa": {"exists": False, "parent_domain": domain, "record": None, "primary_ns": None, "admin_email": None}, + "spf": {"exists": False, "record": None}, + "dkim": {"exists": False, "records": []}, + "dmarc": {"exists": False, "record": None}, + "subdomains": {"subdomains": [], "cname_records": {}, "has_wildcard_dns": False, "hosting_provider": None}, + "http_redirect": {"http_accessible": False, "redirects_to_https": False, "final_url": None, "error": str(e), "redirect_chain": []}, + "sri": {"sri_enabled": False, "total_external_resources": 0, "resources_with_sri": 0, "sri_coverage_percentage": 0, "missing_sri_count": 0, "sri_algorithms_used": [], "error": "Domain analysis failed"} + } + results.append(error_result) + + # Write results to CSV + with open(output_file, 'w', newline='') as f: + writer = csv.writer(f) + # Write header + writer.writerow([ + 'Domain', + 'Timestamp', + 'Parent Domain', + 'SOA Exists', + 'SOA Record', + 'Primary NS', + 'Admin Email', + 'SPF Exists', + 'SPF Record', + 'DKIM Exists', + 'DKIM Records', + 'DMARC Exists', + 'DMARC Record', + 'Discovered Subdomains', + 'CNAME Records', + 'Has Wildcard DNS', + 'Hosting Provider', + 'HTTP Accessible', + 'Redirects to HTTPS', + 'Final URL', + 'Redirect Chain', + 'HTTP Error', + 'SRI Enabled', + 'Total External Resources', + 'Resources With SRI', + 'SRI Coverage %', + 'Missing SRI Count', + 'SRI Algorithms Used', + 'SRI Error' + ]) + + # Write results + for r in results: + writer.writerow([ + r['domain'], + r['timestamp'], + r['soa']['parent_domain'], + r['soa']['exists'], + r['soa'].get('record'), + r['soa'].get('primary_ns'), + r['soa'].get('admin_email'), + r['spf']['exists'], + r['spf'].get('record'), + r['dkim']['exists'], + ';'.join([f"{rec['selector']}:{rec['record']}" for rec in r['dkim']['records']]) if r['dkim']['records'] else '', + r['dmarc']['exists'], + r['dmarc'].get('record'), + ','.join(r['subdomains']['subdomains']), + ','.join([f"{k}:{v}" for k, v in r['subdomains']['cname_records'].items()]), + r['subdomains']['has_wildcard_dns'], + r['subdomains']['hosting_provider'], + r['http_redirect']['http_accessible'], + r['http_redirect']['redirects_to_https'], + r['http_redirect']['final_url'], + ' -> '.join(r['http_redirect'].get('redirect_chain', [])), + r['http_redirect']['error'], + r['sri']['sri_enabled'], + r['sri']['total_external_resources'], + r['sri']['resources_with_sri'], + r['sri']['sri_coverage_percentage'], + r['sri']['missing_sri_count'], + ','.join(r['sri']['sri_algorithms_used']) if r['sri']['sri_algorithms_used'] else '', + r['sri']['error'] + ]) + + # Optionally write filtered subdomains to a separate CSV + if filtered_subdomains_file: + with open(filtered_subdomains_file, 'w', newline='') as f: + writer = csv.writer(f) + writer.writerow(['Domain', 'Filtered Subdomains']) + for r in results: + filtered = r.get('subdomains', {}).get('filtered_subdomains', []) + if filtered: + writer.writerow([r['domain'], ','.join(filtered)]) diff --git a/domain_security_analyzer/cli.py b/domain_security_analyzer/cli.py new file mode 100644 index 0000000..db97664 --- /dev/null +++ b/domain_security_analyzer/cli.py @@ -0,0 +1,148 @@ +"""Command-line interface for the domain security analyzer.""" + +import argparse +import os +import platform +import sys + +from .__version__ import __version__ + +USAGE_EPILOG = """\ +examples: + domain-analyzer domains.txt report.csv + domain-analyzer domains.txt report.csv 20 + domain-analyzer domains.txt report.csv --filtered-subdomains-file filtered.csv +""" + + +def check_required_modules(): + """Verify runtime dependencies and print install guidance if any are missing.""" + missing_modules = [] + + try: + import dns.resolver # noqa: F401 + except ImportError: + missing_modules.append('dnspython') + + try: + import requests # noqa: F401 + except ImportError: + missing_modules.append('requests') + + try: + from bs4 import BeautifulSoup # noqa: F401 + except ImportError: + missing_modules.append('beautifulsoup4') + + if missing_modules: + print("ERROR: Missing required Python packages!") + print("\nPlease install the following packages:") + for module in missing_modules: + print(f" - {module}") + + print("\nInstallation command:") + print(f" pip install {' '.join(missing_modules)}") + print("\nOr if using pip3:") + print(f" pip3 install {' '.join(missing_modules)}") + print("\nIf using a virtual environment, activate it first and then run the pip command.") + + if 'beautifulsoup4' in missing_modules: + print("\nNote: beautifulsoup4 is required for SRI (Subresource Integrity) analysis") + sys.exit(1) + + +def _configure_windows_console(): + """Best-effort console setup for color and UTF-8 output on Windows.""" + if platform.system() != 'Windows': + return + try: + import colorama + colorama.init() # Initialize colorama for Windows color support + except ImportError: + pass # colorama not installed, colors won't work + + # Try to set console to UTF-8 mode + try: + os.system('chcp 65001 > nul') + except Exception: + pass + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + prog='domain-analyzer', + description='Analyze domain security configurations: DNS, email ' + 'authentication (SPF/DKIM/DMARC), subdomain discovery, and ' + 'Subresource Integrity (SRI) scanning.', + epilog=USAGE_EPILOG, + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + parser.add_argument('input_file', help='Text file with one domain per line') + parser.add_argument('output_file', help='Output CSV path') + parser.add_argument( + 'max_workers', nargs='?', type=int, default=None, + help='Number of parallel workers (default: OS-dependent)', + ) + parser.add_argument( + '--include-wildcard-matches', action='store_true', + help='Include subdomains whose DNS answers match the wildcard baseline', + ) + parser.add_argument( + '--filtered-subdomains-file', metavar='PATH', default=None, + help='Write subdomains excluded by wildcard filtering to a separate CSV', + ) + parser.add_argument( + '--version', action='version', version=f'%(prog)s {__version__}', + ) + return parser + + +def main(argv=None): + _configure_windows_console() + + parser = build_parser() + args = parser.parse_args(argv) + + # Check dependencies only after argparse has handled --help/--version so + # those work even in a minimal environment. + check_required_modules() + from .analyzer import analyze_domains_from_file + + default_workers = min(10, (os.cpu_count() or 4) * 2) + max_workers = args.max_workers or default_workers + + input_file = os.path.normpath(args.input_file) + output_file = os.path.normpath(args.output_file) + filtered_subdomains_file = ( + os.path.normpath(args.filtered_subdomains_file) + if args.filtered_subdomains_file else None + ) + + print("\nStarting domain analysis:") + print(f"Operating System: {platform.system()} {platform.release()}") + print(f"Input file: {input_file}") + print(f"Output file: {output_file}") + print(f"Workers: {max_workers}") + if args.include_wildcard_matches: + print("Include wildcard-matched subdomains: True") + if filtered_subdomains_file: + print(f"Filtered subdomains file: {filtered_subdomains_file}") + print("") + + try: + analyze_domains_from_file( + input_file, + output_file, + max_workers, + include_wildcard_matches=args.include_wildcard_matches, + filtered_subdomains_file=filtered_subdomains_file, + ) + except KeyboardInterrupt: + print("\nAnalysis interrupted by user. Partial results may have been saved.") + except Exception as e: + print(f"\nError during analysis: {str(e)}") + sys.exit(1) + + +if __name__ == "__main__": + main() diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..cc6fa49 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,69 @@ +[build-system] +requires = ["setuptools>=61.0", "wheel"] +build-backend = "setuptools.build_meta" + +[project] +name = "domain-security-analyzer" +dynamic = ["version"] +description = "Analyze domain security configurations: DNS, email authentication (SPF/DKIM/DMARC), subdomain discovery, and Subresource Integrity (SRI) scanning." +readme = "README.md" +requires-python = ">=3.7" +license = { text = "MIT" } +authors = [{ name = "CallMarcus" }] +keywords = [ + "security", + "dns", + "sri", + "domain-analysis", + "cybersecurity", + "spf", + "dkim", + "dmarc", + "securityscorecard", +] +classifiers = [ + "Development Status :: 5 - Production/Stable", + "Intended Audience :: System Administrators", + "Intended Audience :: Information Technology", + "Topic :: Security", + "Topic :: Internet :: Name Service (DNS)", + "Topic :: System :: Networking :: Monitoring", + "License :: OSI Approved :: MIT License", + "Operating System :: OS Independent", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.7", + "Programming Language :: Python :: 3.8", + "Programming Language :: Python :: 3.9", + "Programming Language :: Python :: 3.10", + "Programming Language :: Python :: 3.11", +] +dependencies = [ + "dnspython>=2.4.0", + "requests>=2.28.0", + "beautifulsoup4>=4.11.0", +] + +[project.optional-dependencies] +dev = [ + "pytest>=7.0", + "build", + "twine", +] + +[project.urls] +Homepage = "https://github.com/CallMarcus/domain-security-analyzer" +Repository = "https://github.com/CallMarcus/domain-security-analyzer" +Issues = "https://github.com/CallMarcus/domain-security-analyzer/issues" +Changelog = "https://github.com/CallMarcus/domain-security-analyzer/blob/main/CHANGELOG.md" + +[project.scripts] +domain-analyzer = "domain_security_analyzer.cli:main" + +[tool.setuptools] +py-modules = ["domain_analyzer"] + +[tool.setuptools.packages.find] +include = ["domain_security_analyzer*"] + +[tool.setuptools.dynamic] +version = { attr = "domain_security_analyzer.__version__.__version__" } diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_analyzer.py b/tests/test_analyzer.py new file mode 100644 index 0000000..90e6b88 --- /dev/null +++ b/tests/test_analyzer.py @@ -0,0 +1,118 @@ +"""Unit tests for DomainAnalyzer pure logic (no network required).""" + +import csv + +import pytest + +from domain_security_analyzer import DomainAnalyzer +from domain_security_analyzer import analyzer as analyzer_mod + + +@pytest.fixture +def analyzer(): + return DomainAnalyzer() + + +@pytest.mark.parametrize( + "domain,expected", + [ + ("example.com", "example.com"), + ("www.example.com", "example.com"), + ("a.b.example.com", "example.com"), + ], +) +def test_get_parent_domain(analyzer, domain, expected): + assert analyzer.get_parent_domain(domain) == expected + + +@pytest.mark.parametrize( + "url,domain,expected", + [ + ("https://cdn.other.com/a.js", "example.com", True), + ("https://example.com/a.js", "example.com", False), + ("https://www.example.com/a.js", "example.com", False), + ("/local/a.js", "example.com", False), # relative -> not external + ("", "example.com", False), + ], +) +def test_is_external_resource(analyzer, url, domain, expected): + assert analyzer._is_external_resource(url, domain) is expected + + +@pytest.mark.parametrize( + "integrity,expected", + [ + ("sha256-abc", "sha256"), + ("sha384-abc", "sha384"), + ("sha512-abc", "sha512"), + ("md5-abc", "unknown"), + (None, None), + ], +) +def test_extract_hash_algorithm(analyzer, integrity, expected): + assert analyzer._extract_hash_algorithm(integrity) == expected + + +def test_check_sri_counts_and_coverage(analyzer): + html = """ + + + + + + + """ + result = analyzer.check_sri("example.com", html) + assert result["total_external_resources"] == 3 # local.js excluded + assert result["resources_with_sri"] == 1 + assert result["missing_sri_count"] == 2 + assert result["sri_coverage_percentage"] == pytest.approx(33.3, abs=0.1) + assert result["sri_enabled"] is True + assert result["sri_algorithms_used"] == ["sha384"] + assert result["error"] is None + + +def test_check_sri_empty_html(analyzer): + result = analyzer.check_sri("example.com", "") + assert result["error"] == "No HTML content available" + assert result["total_external_resources"] == 0 + + +def test_analyze_domains_from_file_writes_29_columns(tmp_path, monkeypatch): + """End-to-end CSV writing with analyze_domain stubbed (no network).""" + canned = { + "domain": "example.com", + "timestamp": "2026-06-21T00:00:00", + "soa": {"exists": True, "parent_domain": "example.com", "record": "ns rec", + "primary_ns": "ns", "admin_email": "rec"}, + "spf": {"exists": True, "record": '"v=spf1 -all"'}, + "dkim": {"exists": False, "records": []}, + "dmarc": {"exists": True, "record": "v=DMARC1; p=reject"}, + "subdomains": {"subdomains": ["www.example.com"], "cname_records": {}, + "has_wildcard_dns": False, "hosting_provider": None, + "filtered_subdomains": []}, + "http_redirect": {"http_accessible": True, "redirects_to_https": True, + "final_url": "https://example.com", "error": None, + "redirect_chain": []}, + "sri": {"sri_enabled": False, "total_external_resources": 0, + "resources_with_sri": 0, "sri_coverage_percentage": 0, + "missing_sri_count": 0, "sri_algorithms_used": [], "error": None}, + } + monkeypatch.setattr( + analyzer_mod.DomainAnalyzer, "analyze_domain", + lambda self, domain: dict(canned, domain=domain), + ) + + input_file = tmp_path / "in.txt" + input_file.write_text("example.com\n") + output_file = tmp_path / "out.csv" + + analyzer_mod.analyze_domains_from_file(str(input_file), str(output_file), max_workers=1) + + with open(output_file, newline="") as f: + rows = list(csv.reader(f)) + + assert len(rows[0]) == 29 # header column count + assert rows[0][0] == "Domain" + assert len(rows) == 2 # header + one data row + assert rows[1][0] == "example.com" diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 0000000..648f341 --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,42 @@ +"""Tests for the command-line interface.""" + +import pytest + +from domain_security_analyzer import __version__ +from domain_security_analyzer import cli + + +def test_version_flag(capsys): + with pytest.raises(SystemExit) as exc: + cli.main(["--version"]) + assert exc.value.code == 0 + assert __version__ in capsys.readouterr().out + + +def test_help_flag(capsys): + with pytest.raises(SystemExit) as exc: + cli.main(["--help"]) + assert exc.value.code == 0 + out = capsys.readouterr().out + assert "domain-analyzer" in out + assert "--filtered-subdomains-file" in out + + +def test_parser_parses_positional_and_flags(): + parser = cli.build_parser() + args = parser.parse_args( + ["in.txt", "out.csv", "20", "--include-wildcard-matches", + "--filtered-subdomains-file", "filtered.csv"] + ) + assert args.input_file == "in.txt" + assert args.output_file == "out.csv" + assert args.max_workers == 20 + assert args.include_wildcard_matches is True + assert args.filtered_subdomains_file == "filtered.csv" + + +def test_parser_optional_workers_defaults_none(): + parser = cli.build_parser() + args = parser.parse_args(["in.txt", "out.csv"]) + assert args.max_workers is None + assert args.include_wildcard_matches is False diff --git a/tests/test_packaging.py b/tests/test_packaging.py new file mode 100644 index 0000000..ea1da96 --- /dev/null +++ b/tests/test_packaging.py @@ -0,0 +1,24 @@ +"""Smoke tests for package structure and public API.""" + +import importlib + + +def test_package_exposes_public_api(): + pkg = importlib.import_module("domain_security_analyzer") + assert hasattr(pkg, "DomainAnalyzer") + assert hasattr(pkg, "analyze_domains_from_file") + assert isinstance(pkg.__version__, str) + assert pkg.__version__.count(".") >= 2 + + +def test_version_matches_module(): + from domain_security_analyzer import __version__ + from domain_security_analyzer.__version__ import __version__ as mod_version + assert __version__ == mod_version + + +def test_legacy_shim_reexports_analyzer(): + # Backward compatibility: `from domain_analyzer import DomainAnalyzer` + shim = importlib.import_module("domain_analyzer") + from domain_security_analyzer import DomainAnalyzer + assert shim.DomainAnalyzer is DomainAnalyzer