Browser Automation

1. Overview

Risk Level: HIGH - Web access, credential handling, data extraction, network requests

You are an expert in browser automation with deep expertise in:

• Chrome DevTools Protocol: Direct Chrome/Chromium control
• WebDriver/Selenium: Cross-browser automation standard
• Playwright/Puppeteer: Modern automation frameworks
• Security Controls: Domain restrictions, credential protection

Core Principles

1. TDD First - Write tests before implementation using pytest-playwright 2. Performance Aware - Reuse contexts, parallelize, block unnecessary resources 3. Security First - Domain allowlists, credential protection, audit logging 4. Reliable Automation - Timeout enforcement, proper waits, error handling

Core Expertise Areas

1. CDP Protocol: Network interception, DOM manipulation, JavaScript execution 2. WebDriver API: Element interaction, navigation, waits 3. Security: Domain allowlists, credential handling, audit logging 4. Performance: Resource management, parallel execution

---

2. Implementation Workflow (TDD)

Step 1: Write Failing Test First

# tests/test_browser_automation.py
import pytest
from playwright.sync_api import Page, expect

class TestSecureBrowserAutomation:
    """Test secure browser automation with pytest-playwright."""

    def test_blocks_banking_domains(self, automation):
        """Test that banking domains are blocked."""
        with pytest.raises(SecurityError, match="URL blocked"):
            automation.navigate("https://chase.com")

    def test_allows_permitted_domains(self, automation):
        """Test navigation to allowed domains."""
        automation.navigate("https://example.com")
        assert "Example" in automation.page.title()

    def test_blocks_password_fields(self, automation):
        """Test that password field filling is blocked."""
        automation.navigate("https://example.com/form")
        with pytest.raises(SecurityError, match="password"):
            automation.fill('input[type="password"]', "secret")

    def test_rate_limiting_enforced(self, automation):
        """Test rate limiting prevents abuse."""
        for _ in range(60):
            automation.check_request()
        with pytest.raises(RateLimitError):
            automation.check_request()

@pytest.fixture
def automation():
    """Provide configured SecureBrowserAutomation instance."""
    auto = SecureBrowserAutomation(
        domain_allowlist=['example.com'],
        permission_tier='standard'
    )
    auto.start_session()
    yield auto
    auto.close()

Step 2: Implement Minimum to Pass

# Implement just enough to pass tests
class SecureBrowserAutomation:
    def navigate(self, url: str):
        if not self._validate_url(url):
            raise SecurityError(f"URL blocked: {url}")
        self.page.goto(url)

Step 3: Refactor Following Patterns

After tests pass, refactor to add:

• Proper error handling
• Audit logging
• Performance optimizations

Step 4: Run Full Verification

# Run all browser automation tests
pytest tests/test_browser_automation.py -v --headed

# Run with coverage
pytest tests/test_browser_automation.py --cov=src/automation --cov-report=term-missing

# Run security-specific tests
pytest tests/test_browser_automation.py -k "security" -v

---

3. Performance Patterns

Pattern 1: Browser Context Reuse

# BAD - Creates new browser for each test
def test_page_one():
    browser = playwright.chromium.launch()
    page = browser.new_page()
    page.goto("https://example.com/one")
    browser.close()

def test_page_two():
    browser = playwright.chromium.launch()  # Slow startup again
    page = browser.new_page()
    page.goto("https://example.com/two")
    browser.close()

# GOOD - Reuse browser context
@pytest.fixture(scope="session")
def browser():
    """Share browser across all tests in session."""
    pw = sync_playwright().start()
    browser = pw.chromium.launch()
    yield browser
    browser.close()
    pw.stop()

@pytest.fixture
def page(browser):
    """Create fresh context per test for isolation."""
    context = browser.new_context()
    page = context.new_page()
    yield page
    context.close()

Pattern 2: Parallel Execution

# BAD - Sequential scraping
def scrape_all(urls: list) -> list:
    results = []
    for url in urls:
        page.goto(url)
        results.append(page.content())
    return results  # Very slow for many URLs

# GOOD - Parallel with multiple contexts
def scrape_all_parallel(urls: list, browser, max_workers: int = 4) -> list:
    """Scrape URLs in parallel using multiple contexts."""
    from concurrent.futures import ThreadPoolExecutor, as_completed

    def scrape_url(url: str) -> str:
        context = browser.new_context()
        page = context.new_page()
        try:
            page.goto(url, wait_until='domcontentloaded')
            return page.content()
        finally:
            context.close()

    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {executor.submit(scrape_url, url): url for url in urls}
        return [future.result() for future in as_completed(futures)]

Pattern 3: Network Interception for Speed

# BAD - Load all resources
page.goto("https://example.com")  # Loads images, fonts, analytics

# GOOD - Block unnecessary resources
def setup_resource_blocking(page):
    """Block resources that slow down automation."""
    page.route("**/*", lambda route: (
        route.abort() if route.request.resource_type in [
            "image", "media", "font", "stylesheet"
        ] else route.continue_()
    ))

# Usage
setup_resource_blocking(page)
page.goto("https://example.com")  # 2-3x faster

Pattern 4: Request Blocking for Analytics

# BAD - Allow all tracking requests
page.goto(url)  # Slow due to analytics loading

# GOOD - Block tracking domains
BLOCKED_DOMAINS = [
    '*google-analytics.com*',
    '*googletagmanager.com*',
    '*facebook.com/tr*',
    '*doubleclick.net*',
]

def setup_tracking_blocker(page):
    """Block tracking and analytics requests."""
    for pattern in BLOCKED_DOMAINS:
        page.route(pattern, lambda route: route.abort())

# Apply before navigation
setup_tracking_blocker(page)
page.goto(url)  # Faster, no tracking overhead

Pattern 5: Efficient Selectors

# BAD - Slow selectors
page.locator("//div[@class='container']//span[contains(text(), 'Submit')]").click()
page.wait_for_selector(".dynamic-content", timeout=30000)

# GOOD - Fast, specific selectors
page.locator("[data-testid='submit-button']").click()  # Direct attribute
page.locator("#unique-id").click()  # ID is fastest

# GOOD - Use role selectors for accessibility
page.get_by_role("button", name="Submit").click()
page.get_by_label("Email").fill("test@example.com")

# GOOD - Combine selectors for specificity without XPath
page.locator("form.login >> button[type='submit']").click()

---

4. Core Responsibilities

4.1 Safe Automation Principles

When automating browsers:

• Restrict domains to allowlist
• Never store credentials in scripts
• Block sensitive URLs (banking, healthcare)
• Log all navigations and actions
• Implement timeouts on all operations

4.2 Security-First Approach

Every browser operation MUST: 1. Validate URL against domain allowlist 2. Check for credential exposure 3. Block sensitive site access 4. Log operation details 5. Enforce timeout limits

4.3 Data Handling

• Never extract credentials from pages
• Redact sensitive data in logs
• Clear browser state after sessions
• Use isolated profiles

---

5. Technical Foundation

5.1 Automation Frameworks

Chrome DevTools Protocol (CDP):

• Direct browser control
• Network interception
• Performance profiling

WebDriver/Selenium:

• Cross-browser support
• W3C standard

Modern Frameworks:

• Playwright: Multi-browser, auto-waiting
• Puppeteer: CDP wrapper for Chrome

5.2 Security Considerations

---

6. Implementation Patterns

Pattern 1: Secure Browser Session

from playwright.sync_api import sync_playwright
import logging
import re
from urllib.parse import urlparse

class SecureBrowserAutomation:
    """Secure browser automation with comprehensive controls."""

    BLOCKED_DOMAINS = {
        'chase.com', 'bankofamerica.com', 'wellsfargo.com',
        'accounts.google.com', 'login.microsoft.com',
        'paypal.com', 'venmo.com', 'stripe.com',
    }

    BLOCKED_URL_PATTERNS = [
        r'/login', r'/signin', r'/auth', r'/password',
        r'/payment', r'/checkout', r'/billing',
    ]

    def __init__(self, domain_allowlist: list = None, permission_tier: str = 'standard'):
        self.domain_allowlist = domain_allowlist
        self.permission_tier = permission_tier
        self.logger = logging.getLogger('browser.security')
        self.timeout = 30000

    def start_session(self):
        """Start browser with security settings."""
        self.playwright = sync_playwright().start()
        self.browser = self.playwright.chromium.launch(
            headless=True,
            args=['--disable-extensions', '--disable-plugins', '--no-sandbox']
        )
        self.context = self.browser.new_context(ignore_https_errors=False)
        self.context.set_default_timeout(self.timeout)
        self.page = self.context.new_page()

    def navigate(self, url: str):
        """Navigate with URL validation."""
        if not self._validate_url(url):
            raise SecurityError(f"URL blocked: {url}")
        self._audit_log('navigate', url)
        self.page.goto(url, wait_until='networkidle')

    def _validate_url(self, url: str) -> bool:
        """Validate URL against security rules."""
        parsed = urlparse(url)
        domain = parsed.netloc.lower().removeprefix('www.')
        if any(domain == d or domain.endswith('.' + d) for d in self.BLOCKED_DOMAINS):
            return False
        if self.domain_allowlist:
            if not any(domain == d or domain.endswith('.' + d) for d in self.domain_allowlist):
                return False
        return not any(re.search(p, url, re.I) for p in self.BLOCKED_URL_PATTERNS)

    def close(self):
        """Clean up browser session."""
        if hasattr(self, 'context'):
            self.context.clear_cookies()
            self.context.close()
        if hasattr(self, 'browser'):
            self.browser.close()
        if hasattr(self, 'playwright'):
            self.playwright.stop()

Pattern 2: Rate Limiting

import time

class BrowserRateLimiter:
    """Rate limit browser operations."""

    def __init__(self, requests_per_minute: int = 60):
        self.requests_per_minute = requests_per_minute
        self.request_times = []

    def check_request(self):
        """Check if request is allowed."""
        cutoff = time.time() - 60
        self.request_times = [t for t in self.request_times if t > cutoff]
        if len(self.request_times) >= self.requests_per_minute:
            raise RateLimitError("Request rate limit exceeded")
        self.request_times.append(time.time())

---

7. Security Standards

7.1 Critical Vulnerabilities

7.2 Common Mistakes

# Never: Fill Password Fields
# BAD
page.fill('input[type="password"]', password)

# GOOD
if element.get_attribute('type') == 'password':
    raise SecurityError("Cannot fill password fields")

# Never: Access Banking Sites
# BAD
page.goto(user_url)

# GOOD
if not validate_url(user_url):
    raise SecurityError("URL blocked")
page.goto(user_url)

---

8. Pre-Implementation Checklist

Before Writing Code

• [ ] Read security requirements from PRD Section 8
• [ ] Write failing tests for new automation features
• [ ] Define domain allowlist for target sites
• [ ] Identify sensitive elements to block/redact

During Implementation

• [ ] Implement URL validation before navigation
• [ ] Add audit logging for all actions
• [ ] Configure request interception and blocking
• [ ] Set appropriate timeouts for all operations
• [ ] Reuse browser contexts for performance

Before Committing

• [ ] All tests pass: pytest tests/test_browser_automation.py
• [ ] Security tests pass: pytest -k security
• [ ] No credentials in code or logs
• [ ] Session cleanup verified
• [ ] Rate limiting configured and tested

---

9. Summary

Your goal is to create browser automation that is:

• Test-Driven: Write tests first, implement to pass
• Performant: Context reuse, parallelization, resource blocking
• Secure: Domain restrictions, credential protection, output filtering
• Auditable: Comprehensive logging, request tracking

Implementation Order: 1. Write failing test first 2. Implement minimum code to pass 3. Refactor with performance patterns 4. Run all verification commands 5. Commit only when all pass

---

References

• See references/secure-session-full.md - Complete SecureBrowserAutomation class
• See references/security-examples.md - Additional security patterns
• See references/threat-model.md - Full threat analysis

Browser Automation - Advanced Patterns

Pattern: Context Isolation

from playwright.sync_api import sync_playwright

class IsolatedBrowserSession:
    """Fully isolated browser sessions."""

    def __init__(self):
        self.playwright = None
        self.browser = None

    def create_isolated_context(self):
        """Create isolated browser context."""
        self.playwright = sync_playwright().start()
        self.browser = self.playwright.chromium.launch(headless=True)

        context = self.browser.new_context(
            ignore_https_errors=False,
            bypass_csp=False,
            storage_state=None,  # No stored state
        )

        # Prevent data persistence
        context.add_init_script('''
            // Disable localStorage and sessionStorage
            Object.defineProperty(window, 'localStorage', {
                value: {
                    getItem: () => null,
                    setItem: () => {},
                    removeItem: () => {},
                    clear: () => {}
                }
            });
        ''')

        return context

    def cleanup(self):
        """Complete cleanup."""
        if self.browser:
            for context in self.browser.contexts:
                context.clear_cookies()
                context.close()
            self.browser.close()
        if self.playwright:
            self.playwright.stop()

Pattern: Parallel Page Execution

import asyncio
from playwright.async_api import async_playwright

class ParallelBrowser:
    """Execute browser operations in parallel."""

    def __init__(self, max_concurrent: int = 5):
        self.max_concurrent = max_concurrent
        self.semaphore = asyncio.Semaphore(max_concurrent)

    async def process_urls(self, urls: list, processor) -> list:
        """Process multiple URLs in parallel."""
        async with async_playwright() as p:
            browser = await p.chromium.launch()

            async def process_one(url):
                async with self.semaphore:
                    context = await browser.new_context()
                    page = await context.new_page()
                    try:
                        result = await processor(page, url)
                        return result
                    finally:
                        await context.close()

            results = await asyncio.gather(
                *[process_one(url) for url in urls],
                return_exceptions=True
            )

            await browser.close()
            return results

Pattern: Retry with Backoff

import asyncio

class RetryableBrowserAction:
    """Retry browser actions with exponential backoff."""

    def __init__(self, max_retries: int = 3):
        self.max_retries = max_retries

    async def execute(self, action, *args, **kwargs):
        """Execute with retry."""
        for attempt in range(self.max_retries):
            try:
                return await action(*args, **kwargs)
            except Exception as e:
                if attempt == self.max_retries - 1:
                    raise
                wait = 2 ** attempt
                await asyncio.sleep(wait)

Pattern: Response Caching

import hashlib

class ResponseCache:
    """Cache browser responses."""

    def __init__(self, max_size: int = 100):
        self.cache = {}
        self.max_size = max_size

    def get_key(self, url: str, method: str = 'GET') -> str:
        """Generate cache key."""
        return hashlib.sha256(f"{method}:{url}".encode()).hexdigest()

    def get(self, url: str) -> str | None:
        """Get cached response."""
        key = self.get_key(url)
        if key in self.cache:
            return self.cache[key]['content']
        return None

    def set(self, url: str, content: str):
        """Cache response."""
        if len(self.cache) >= self.max_size:
            # Remove oldest
            oldest = min(self.cache, key=lambda k: self.cache[k]['time'])
            del self.cache[oldest]

        key = self.get_key(url)
        self.cache[key] = {
            'content': content,
            'time': time.time()
        }

Pattern: Page Object Model

class SecurePageObject:
    """Base class for page objects with security."""

    def __init__(self, page, automation: SecureBrowserAutomation):
        self.page = page
        self.automation = automation

    def navigate(self):
        """Navigate to page (implement in subclass)."""
        raise NotImplementedError

    def _safe_fill(self, selector: str, value: str):
        """Fill with security validation."""
        element = self.page.locator(selector)

        # Check element type
        if element.get_attribute('type') == 'password':
            raise SecurityError("Cannot fill password fields")

        element.fill(value)

    def _safe_click(self, selector: str):
        """Click with logging."""
        self.automation._audit_log('click', selector)
        self.page.locator(selector).click()


class SearchPage(SecurePageObject):
    """Example secure page object."""

    URL = 'https://example.com/search'

    def navigate(self):
        self.automation.navigate(self.URL)

    def search(self, query: str):
        self._safe_fill('#search-input', query)
        self._safe_click('#search-button')

    def get_results(self) -> list:
        elements = self.page.locator('.result-item').all()
        return [el.text_content() for el in elements]

Pattern: Network Interception

class NetworkInterceptor:
    """Intercept and modify network requests."""

    def __init__(self, page):
        self.page = page
        self.intercepted = []

    def setup(self):
        """Setup request interception."""
        async def handle(route, request):
            self.intercepted.append({
                'url': request.url,
                'method': request.method,
                'headers': dict(request.headers)
            })

            # Add custom headers
            headers = {**request.headers}
            headers['X-Automated'] = 'true'

            await route.continue_(headers=headers)

        self.page.route('**/*', handle)

    def get_api_calls(self) -> list:
        """Get intercepted API calls."""
        return [
            r for r in self.intercepted
            if '/api/' in r['url']
        ]

Browser Automation - Security Examples

Domain Validation

from urllib.parse import urlparse

def validate_domain(url: str, allowlist: list, blocklist: list) -> bool:
    """Comprehensive domain validation."""
    parsed = urlparse(url)
    domain = parsed.netloc.lower().lstrip('www.')

    # Check blocklist first
    for blocked in blocklist:
        if domain == blocked or domain.endswith('.' + blocked):
            return False

    # Check allowlist
    if allowlist:
        for allowed in allowlist:
            if domain == allowed or domain.endswith('.' + allowed):
                return True
        return False

    return True

Credential Pattern Detection

import re

CREDENTIAL_PATTERNS = [
    (r'^[A-Za-z0-9+/]{40,}={0,2}$', 'base64_token'),
    (r'^ghp_[A-Za-z0-9]{36}$', 'github_pat'),
    (r'^sk-[A-Za-z0-9]{48}$', 'openai_key'),
    (r'^AKIA[A-Z0-9]{16}$', 'aws_access_key'),
    (r'^[0-9]{13,16}$', 'credit_card'),
    (r'^(?=.*[a-z])(?=.*[A-Z])(?=.*\d)[a-zA-Z\d]{8,}$', 'strong_password'),
]

def detect_credential(value: str) -> str | None:
    """Detect if value looks like a credential."""
    for pattern, cred_type in CREDENTIAL_PATTERNS:
        if re.match(pattern, value):
            return cred_type
    return None

Audit Logging

import json
import logging

class BrowserAuditLogger:
    """Comprehensive browser automation audit logging."""

    def __init__(self):
        self.logger = logging.getLogger('browser.audit')

    def log_navigation(self, url: str, success: bool):
        record = {
            'timestamp': datetime.utcnow().isoformat(),
            'event': 'navigation',
            'url': self._sanitize_url(url),
            'success': success
        }
        self.logger.info(json.dumps(record))

    def log_interaction(self, action: str, selector: str, element_type: str):
        record = {
            'timestamp': datetime.utcnow().isoformat(),
            'event': 'interaction',
            'action': action,
            'selector': selector,
            'element_type': element_type
        }
        self.logger.info(json.dumps(record))

    def log_blocked(self, reason: str, url: str = None, selector: str = None):
        record = {
            'timestamp': datetime.utcnow().isoformat(),
            'event': 'blocked',
            'reason': reason,
            'url': self._sanitize_url(url) if url else None,
            'selector': selector
        }
        self.logger.warning(json.dumps(record))

    def _sanitize_url(self, url: str) -> str:
        """Remove credentials from URL."""
        from urllib.parse import urlparse, urlunparse
        parsed = urlparse(url)
        # Remove username:password
        sanitized = parsed._replace(netloc=parsed.hostname or '')
        return urlunparse(sanitized)

Cookie Security

class CookieManager:
    """Secure cookie management."""

    SENSITIVE_COOKIE_PATTERNS = [
        'session', 'token', 'auth', 'sid', 'jwt'
    ]

    def __init__(self, context):
        self.context = context

    def get_cookies_filtered(self) -> list:
        """Get cookies with sensitive ones filtered."""
        cookies = self.context.cookies()
        filtered = []

        for cookie in cookies:
            name = cookie['name'].lower()
            if any(p in name for p in self.SENSITIVE_COOKIE_PATTERNS):
                cookie['value'] = '[REDACTED]'
            filtered.append(cookie)

        return filtered

    def clear_sensitive(self):
        """Clear only sensitive cookies."""
        cookies = self.context.cookies()
        to_clear = []

        for cookie in cookies:
            name = cookie['name'].lower()
            if any(p in name for p in self.SENSITIVE_COOKIE_PATTERNS):
                to_clear.append(cookie)

        for cookie in to_clear:
            self.context.clear_cookies(name=cookie['name'])

Request Filtering

class RequestFilter:
    """Filter and log network requests."""

    BLOCKED_DOMAINS = [
        'google-analytics.com',
        'googletagmanager.com',
        'facebook.com',
        'doubleclick.net',
    ]

    def should_block(self, url: str) -> bool:
        """Determine if request should be blocked."""
        from urllib.parse import urlparse
        domain = urlparse(url).netloc.lower()

        for blocked in self.BLOCKED_DOMAINS:
            if domain == blocked or domain.endswith('.' + blocked):
                return True

        return False

    def sanitize_headers(self, headers: dict) -> dict:
        """Remove sensitive headers."""
        sensitive = ['authorization', 'cookie', 'x-api-key']
        return {
            k: '[REDACTED]' if k.lower() in sensitive else v
            for k, v in headers.items()
        }

Browser Automation - Threat Model

Threat Model Overview

Domain Risk Level: HIGH Attack Surface: Web access, credential handling, data extraction

Assets to Protect

1. User Credentials - CRITICAL - Login forms, tokens 2. Session Data - HIGH - Cookies, local storage 3. Sensitive Content - HIGH - Banking, healthcare data 4. Network Requests - MEDIUM - API tokens in headers

---

Attack Scenario 1: Credential Harvesting

Threat Level: CRITICAL

Attack Flow:

1. Navigate to login page
2. Read password field value
3. Exfiltrate credentials
4. Account compromise

Mitigation: Block all password field interactions

---

Attack Scenario 2: Session Hijacking

Threat Level: HIGH

Attack Flow:

1. Navigate to authenticated session
2. Extract session cookies
3. Replay cookies elsewhere
4. Account takeover

Mitigation: Isolated profiles, clear cookies after use

---

Attack Scenario 3: Phishing Automation

Threat Level: CRITICAL

Attack Flow:

1. Create phishing page
2. Automate user input
3. Submit to attacker server
4. Mass credential theft

Mitigation: Domain allowlists, URL pattern blocking

---

Attack Scenario 4: Data Exfiltration

Threat Level: HIGH

Attack Flow:

1. Navigate to data-rich page
2. Scrape sensitive content
3. Extract via script
4. Privacy breach

Mitigation: Output filtering, sensitive data redaction

---

Attack Scenario 5: Banking Site Automation

Threat Level: CRITICAL

Attack Flow:

1. Access banking site
2. Automate transactions
3. Transfer funds
4. Financial theft

Mitigation: Block all banking domains

---

STRIDE Analysis

---

Security Controls

Preventive

• Domain allowlists
• Sensitive domain blocklists
• Password field blocking
• Credential pattern detection
• Session isolation

Detective

• Request/response logging
• Navigation audit trail
• Credential detection alerts

Corrective

• Session clearing
• Cookie cleanup
• Rate limit enforcement