Building Custom Security Tools: From Concept to Implementation#

As a penetration tester and security researcher, I've learned that sometimes the best tool for the job is the one you build yourself. Custom security tools can fill gaps in existing solutions, automate repetitive tasks, and provide tailored functionality for specific environments.

Why Build Custom Tools?#

Advantages of Custom Development#

Tailored functionality: Address specific needs not covered by existing tools
Learning opportunity: Deepen understanding of security concepts
Competitive advantage: Unique tools can set you apart
Community contribution: Share tools to help the security community

When to Build vs. Buy/Use Existing#

Build when:

Existing tools don't meet specific requirements
You need to understand the underlying technology
Integration with custom workflows is required
Learning and skill development is a priority

Use existing when:

Mature, well-tested solutions exist
Time constraints are critical
Maintenance overhead would be significant

Planning Your Security Tool#

Define the Problem#

Start with a clear problem statement:

# Example: Subdomain enumeration tool
PROBLEM = """
Existing subdomain enumeration tools are either:
1. Too slow for large domains
2. Missing specific data sources
3. Don't integrate with our reporting pipeline
"""

SOLUTION = """
Build a fast, modular subdomain enumerator that:
- Uses multiple data sources concurrently
- Provides structured output
- Integrates with existing tools
"""

Architecture Design#

Consider these architectural patterns:

Modular design: Separate concerns into distinct modules
Plugin system: Allow extensibility through plugins
Configuration-driven: Use config files for flexibility
API-first: Design with integration in mind

Development Best Practices#

Code Structure#

# Example project structure
security_tool/
├── src/
│   ├── core/
│   │   ├── __init__.py
│   │   ├── scanner.py
│   │   └── reporter.py
│   ├── modules/
│   │   ├── __init__.py
│   │   ├── dns_enum.py
│   │   └── port_scan.py
│   └── utils/
│       ├── __init__.py
│       └── helpers.py
├── config/
│   └── settings.yaml
├── tests/
├── docs/
└── requirements.txt

Error Handling and Logging#

import logging
import sys
from typing import Optional

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('security_tool.log'),
        logging.StreamHandler(sys.stdout)
    ]
)

logger = logging.getLogger(__name__)

class SecurityToolError(Exception):
    """Base exception for security tool errors"""
    pass

class ScannerError(SecurityToolError):
    """Scanner-specific errors"""
    pass

def safe_scan(target: str) -> Optional[dict]:
    """Safely perform scan with proper error handling"""
    try:
        result = perform_scan(target)
        logger.info(f"Successfully scanned {target}")
        return result
    except ScannerError as e:
        logger.error(f"Scanner error for {target}: {e}")
        return None
    except Exception as e:
        logger.critical(f"Unexpected error for {target}: {e}")
        raise

Configuration Management#

# config/settings.yaml
scanner:
  timeout: 30
  max_threads: 50
  retry_attempts: 3

output:
  format: json
  file: results.json
  verbose: true

modules:
  dns_enum:
    enabled: true
    wordlist: /path/to/wordlist.txt
  port_scan:
    enabled: true
    ports: [80, 443, 8080, 8443]

Example: Building a Subdomain Enumerator#

Core Scanner Class#

import asyncio
import aiohttp
import dns.resolver
from concurrent.futures import ThreadPoolExecutor
from typing import List, Set

class SubdomainEnumerator:
    def __init__(self, domain: str, wordlist: str, threads: int = 50):
        self.domain = domain
        self.wordlist = wordlist
        self.threads = threads
        self.found_subdomains: Set[str] = set()
        
    async def enumerate(self) -> List[str]:
        """Main enumeration method"""
        tasks = []
        
        # DNS brute force
        tasks.append(self.dns_bruteforce())
        
        # Certificate transparency
        tasks.append(self.cert_transparency())
        
        # Search engines
        tasks.append(self.search_engines())
        
        # Run all tasks concurrently
        await asyncio.gather(*tasks)
        
        return sorted(list(self.found_subdomains))
    
    async def dns_bruteforce(self):
        """Brute force DNS subdomains"""
        with open(self.wordlist, 'r') as f:
            subdomains = [line.strip() for line in f]
        
        semaphore = asyncio.Semaphore(self.threads)
        tasks = []
        
        for subdomain in subdomains:
            task = self.check_subdomain(subdomain, semaphore)
            tasks.append(task)
        
        await asyncio.gather(*tasks, return_exceptions=True)
    
    async def check_subdomain(self, subdomain: str, semaphore):
        """Check if subdomain exists"""
        async with semaphore:
            full_domain = f"{subdomain}.{self.domain}"
            try:
                # DNS resolution
                resolver = dns.resolver.Resolver()
                resolver.timeout = 2
                resolver.lifetime = 2
                
                answers = resolver.resolve(full_domain, 'A')
                if answers:
                    self.found_subdomains.add(full_domain)
                    logger.info(f"Found: {full_domain}")
            except Exception:
                pass  # Subdomain doesn't exist or timeout
    
    async def cert_transparency(self):
        """Query certificate transparency logs"""
        url = f"https://crt.sh/?q=%.{self.domain}&output=json"
        
        async with aiohttp.ClientSession() as session:
            try:
                async with session.get(url) as response:
                    if response.status == 200:
                        data = await response.json()
                        for entry in data:
                            name = entry.get('name_value', '')
                            if name and name.endswith(self.domain):
                                self.found_subdomains.add(name)
            except Exception as e:
                logger.error(f"Certificate transparency error: {e}")

Command Line Interface#

import argparse
import asyncio
import json
from pathlib import Path

def main():
    parser = argparse.ArgumentParser(
        description="Advanced Subdomain Enumerator"
    )
    parser.add_argument(
        "domain", 
        help="Target domain to enumerate"
    )
    parser.add_argument(
        "-w", "--wordlist",
        default="wordlist.txt",
        help="Wordlist file for brute force"
    )
    parser.add_argument(
        "-t", "--threads",
        type=int,
        default=50,
        help="Number of threads"
    )
    parser.add_argument(
        "-o", "--output",
        help="Output file (JSON format)"
    )
    parser.add_argument(
        "-v", "--verbose",
        action="store_true",
        help="Verbose output"
    )
    
    args = parser.parse_args()
    
    if args.verbose:
        logging.getLogger().setLevel(logging.DEBUG)
    
    # Validate wordlist
    if not Path(args.wordlist).exists():
        logger.error(f"Wordlist not found: {args.wordlist}")
        return 1
    
    # Run enumeration
    enumerator = SubdomainEnumerator(
        domain=args.domain,
        wordlist=args.wordlist,
        threads=args.threads
    )
    
    try:
        subdomains = asyncio.run(enumerator.enumerate())
        
        results = {
            "domain": args.domain,
            "subdomains": subdomains,
            "count": len(subdomains)
        }
        
        if args.output:
            with open(args.output, 'w') as f:
                json.dump(results, f, indent=2)
            logger.info(f"Results saved to {args.output}")
        else:
            print(json.dumps(results, indent=2))
            
    except KeyboardInterrupt:
        logger.info("Enumeration interrupted by user")
        return 1
    except Exception as e:
        logger.error(f"Enumeration failed: {e}")
        return 1
    
    return 0

if __name__ == "__main__":
    exit(main())

Testing and Quality Assurance#

Unit Testing#

import unittest
from unittest.mock import patch, MagicMock

class TestSubdomainEnumerator(unittest.TestCase):
    def setUp(self):
        self.enumerator = SubdomainEnumerator(
            domain="example.com",
            wordlist="test_wordlist.txt",
            threads=10
        )
    
    @patch('dns.resolver.Resolver')
    async def test_check_subdomain_exists(self, mock_resolver):
        # Mock DNS resolution
        mock_resolver.return_value.resolve.return_value = [MagicMock()]
        
        semaphore = asyncio.Semaphore(1)
        await self.enumerator.check_subdomain("www", semaphore)
        
        self.assertIn("www.example.com", self.enumerator.found_subdomains)
    
    def test_domain_validation(self):
        # Test domain validation logic
        valid_domains = ["example.com", "sub.example.com"]
        invalid_domains = ["invalid", ""]
        
        for domain in valid_domains:
            self.assertTrue(self.enumerator.is_valid_domain(domain))
        
        for domain in invalid_domains:
            self.assertFalse(self.enumerator.is_valid_domain(domain))

Performance Testing#

import time
import asyncio
from memory_profiler import profile

@profile
def performance_test():
    """Test tool performance with large datasets"""
    start_time = time.time()
    
    enumerator = SubdomainEnumerator(
        domain="example.com",
        wordlist="large_wordlist.txt",
        threads=100
    )
    
    results = asyncio.run(enumerator.enumerate())
    
    end_time = time.time()
    duration = end_time - start_time
    
    print(f"Enumerated {len(results)} subdomains in {duration:.2f} seconds")
    print(f"Rate: {len(results)/duration:.2f} subdomains/second")

Distribution and Maintenance#

Packaging#

# setup.py
from setuptools import setup, find_packages

setup(
    name="subdomain-enumerator",
    version="1.0.0",
    description="Advanced subdomain enumeration tool",
    author="ibrahimsql",
    author_email="ibrahimsql@proton.me",
    packages=find_packages(),
    install_requires=[
        "aiohttp>=3.8.0",
        "dnspython>=2.2.0",
        "pyyaml>=6.0",
    ],
    entry_points={
        "console_scripts": [
            "subenum=subdomain_enumerator.cli:main",
        ],
    },
    classifiers=[
        "Development Status :: 4 - Beta",
        "Intended Audience :: Information Technology",
        "License :: OSI Approved :: MIT License",
        "Programming Language :: Python :: 3.8+",
    ],
)

Documentation#

Create comprehensive documentation:

README.md: Installation and basic usage
API documentation: Detailed function/class documentation
Examples: Real-world usage scenarios
Contributing guidelines: How others can contribute

Security Considerations#

Responsible Development#

Rate limiting: Respect target systems
Error handling: Fail gracefully
Logging: Don't log sensitive information
Permissions: Follow principle of least privilege

Legal and Ethical Guidelines#

Only test systems you own or have permission to test
Include clear usage disclaimers
Follow responsible disclosure practices
Respect rate limits and terms of service

Conclusion#

Building custom security tools is both an art and a science. It requires technical skills, security knowledge, and careful consideration of ethical implications. The key is to start small, iterate quickly, and always prioritize quality and responsibility.

Remember: the best tool is one that solves a real problem efficiently and safely.

Want to see more tool development tutorials? Follow me for updates on my latest security research and open-source projects.