How We Survived Firewall vs Password Manager: What You Need to Know

import requests
import time
import logging
from typing import List, Dict, Optional

# Configure logging for audit trails
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    handlers=[logging.FileHandler("firewall_rules.log"), logging.StreamHandler()]
)
logger = logging.getLogger(__name__)

class PfSenseFirewallClient:
    """Client for pfSense 2.7.0+ API (https://github.com/pfsense/pfsense-api)"""

    def __init__(self, base_url: str, api_key: str, api_secret: str):
        self.base_url = base_url.rstrip("/")
        self.session = requests.Session()
        self.session.auth = (api_key, api_secret)
        self.session.headers.update({"Content-Type": "application/json"})
        self.rate_limit_delay = 0.5  # 2 requests per second to avoid rate limits
        self.max_retries = 3

    def _make_request(self, method: str, endpoint: str, payload: Optional[Dict] = None) -> Dict:
        """Handle rate limiting, retries, and error handling for API requests"""
        url = f"{self.base_url}/api/v1/{endpoint.lstrip('/')}"
        retries = 0

        while retries < self.max_retries:
            try:
                response = self.session.request(method, url, json=payload, timeout=10)
                response.raise_for_status()
                time.sleep(self.rate_limit_delay)
                return response.json()
            except requests.exceptions.HTTPError as e:
                if e.response.status_code == 429:  # Rate limited
                    retry_after = int(e.response.headers.get("Retry-After", 5))
                    logger.warning(f"Rate limited, retrying after {retry_after}s")
                    time.sleep(retry_after)
                    retries += 1
                elif e.response.status_code == 401:
                    logger.error("Authentication failed: Invalid API key/secret")
                    raise
                else:
                    logger.error(f"HTTP error {e.response.status_code}: {e.response.text}")
                    raise
            except requests.exceptions.RequestException as e:
                logger.error(f"Request failed: {str(e)}")
                retries += 1
                time.sleep(2 ** retries)  # Exponential backoff
        raise Exception(f"Max retries exceeded for {method} {url}")

    def add_rule(self, interface: str, action: str, protocol: str, source: str, destination: str, port: Optional[str] = None) -> Dict:
        """Add a stateful firewall rule to pfSense

        Args:
            interface: Firewall interface (e.g., "wan", "lan")
            action: "pass" or "block"
            protocol: "tcp", "udp", "icmp"
            source: CIDR or IP (e.g., "192.168.1.0/24")
            destination: CIDR or IP (e.g., "10.0.0.0/8")
            port: Optional port range (e.g., "443" or "8080-8081")

        Returns:
            API response with rule ID
        """
        # Validate inputs
        if action not in ("pass", "block"):
            raise ValueError(f"Invalid action: {action}. Must be 'pass' or 'block'")
        if protocol not in ("tcp", "udp", "icmp"):
            raise ValueError(f"Invalid protocol: {protocol}")

        rule_payload = {
            "interface": interface,
            "type": action,
            "protocol": protocol,
            "source": source,
            "destination": destination,
            "descr": f"Auto-added rule {time.time()}",
            "disabled": False
        }
        if port:
            rule_payload["destination_port"] = port

        logger.info(f"Adding rule: {action} {protocol} from {source} to {destination}:{port or 'any'}")
        return self._make_request("POST", "/firewall/rule", rule_payload)

    def list_rules(self, interface: Optional[str] = None) -> List[Dict]:
        """List all firewall rules, optionally filtered by interface"""
        rules = self._make_request("GET", "/firewall/rule")
        if interface:
            return [r for r in rules if r.get("interface") == interface]
        return rules

if __name__ == "__main__":
    # Example usage: Add a rule to block SSH from external IPs
    client = PfSenseFirewallClient(
        base_url="https://pfsense.example.com",
        api_key="your_api_key",
        api_secret="your_api_secret"
    )

    try:
        # List existing WAN rules
        wan_rules = client.list_rules(interface="wan")
        logger.info(f"Found {len(wan_rules)} existing WAN rules")

        # Add block rule for SSH from all external IPs
        new_rule = client.add_rule(
            interface="wan",
            action="block",
            protocol="tcp",
            source="0.0.0.0/0",
            destination="192.168.1.0/24",
            port="22"
        )
        logger.info(f"Added new rule with ID: {new_rule.get('id')}")
    except Exception as e:
        logger.error(f"Failed to manage firewall rules: {str(e)}")
        exit(1)

import subprocess
import os
import json
import logging
from typing import Optional, Dict, List

logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
logger = logging.getLogger(__name__)

class BitwardenCIClient:
    """Client for Bitwarden CLI (https://github.com/bitwarden/clients) to retrieve secrets in CI/CD"""

    def __init__(self, client_id: str, client_secret: str, master_password: str):
        self.client_id = client_id
        self.client_secret = client_secret
        self.master_password = master_password
        self.session_token: Optional[str] = None
        self.bw_path = "bw"  # Assumes bw is in PATH, install from https://github.com/bitwarden/clients/releases

    def _run_bw_command(self, args: List[str], input_data: Optional[str] = None) -> str:
        """Run a Bitwarden CLI command and handle errors"""
        cmd = [self.bw_path] + args
        try:
            result = subprocess.run(
                cmd,
                input=input_data.encode() if input_data else None,
                capture_output=True,
                timeout=30
            )
            if result.returncode != 0:
                error_msg = result.stderr.decode().strip()
                logger.error(f"bw command failed: {error_msg}")
                raise Exception(f"Bitwarden CLI error: {error_msg}")
            return result.stdout.decode().strip()
        except subprocess.TimeoutExpired:
            logger.error(f"bw command timed out: {' '.join(cmd)}")
            raise

    def login(self) -> None:
        """Login to Bitwarden using client credentials flow (CI/CD optimized)"""
        # Set environment variables for non-interactive login
        os.environ["BW_CLIENTID"] = self.client_id
        os.environ["BW_CLIENTSECRET"] = self.client_secret

        # Unlock vault with master password
        logger.info("Logging into Bitwarden vault...")
        output = self._run_bw_command(
            ["login", "--apikey"],
            input_data=f"{self.master_password}\n"
        )
        logger.info("Login successful")

        # Get session token for subsequent requests
        self.session_token = self._run_bw_command(["unlock", "--raw"], input_data=f"{self.master_password}\n")
        os.environ["BW_SESSION"] = self.session_token
        logger.info("Vault unlocked, session token set")

    def get_secret(self, item_id: str) -> Dict:
        """Retrieve a secret by Bitwarden item ID"""
        if not self.session_token:
            raise Exception("Not logged in. Call login() first.")

        logger.info(f"Retrieving secret with ID: {item_id}")
        output = self._run_bw_command(["get", "item", item_id])
        try:
            item = json.loads(output)
            # Extract password from item
            password = item.get("login", {}).get("password")
            if not password:
                raise Exception(f"No password found in item {item_id}")
            return {
                "id": item["id"],
                "name": item["name"],
                "password": password,
                "username": item.get("login", {}).get("username")
            }
        except json.JSONDecodeError:
            raise Exception(f"Failed to parse Bitwarden item response: {output}")

    def list_secrets(self, collection_id: Optional[str] = None) -> List[Dict]:
        """List all secrets, optionally filtered by collection ID"""
        args = ["list", "items"]
        if collection_id:
            args.extend(["--collectionid", collection_id])
        output = self._run_bw_command(args)
        try:
            items = json.loads(output)
            return [{"id": i["id"], "name": i["name"]} for i in items]
        except json.JSONDecodeError:
            raise Exception(f"Failed to parse item list: {output}")

    def sync_vault(self) -> None:
        """Sync vault with Bitwarden servers to get latest secrets"""
        logger.info("Syncing Bitwarden vault...")
        self._run_bw_command(["sync"])
        logger.info("Vault sync complete")

if __name__ == "__main__":
    # Example usage: Retrieve database credentials in CI pipeline
    # Store these in CI/CD secrets, not hardcoded!
    client = BitwardenCIClient(
        client_id=os.environ.get("BITWARDEN_CLIENT_ID"),
        client_secret=os.environ.get("BITWARDEN_CLIENT_SECRET"),
        master_password=os.environ.get("BITWARDEN_MASTER_PASSWORD")
    )

    try:
        client.login()
        client.sync_vault()

        # List all secrets in the "Production" collection
        production_secrets = client.list_secrets(collection_id="prod_collection_123")
        logger.info(f"Found {len(production_secrets)} production secrets")

        # Retrieve database password
        db_secret = client.get_secret(item_id="db_creds_456")
        logger.info(f"Retrieved secret: {db_secret['name']}")
        # Use db_secret["password"] to connect to database
    except Exception as e:
        logger.error(f"CI/CD secret retrieval failed: {str(e)}")
        exit(1)

import subprocess
import requests
import time
import json
import statistics
from typing import Dict, List, Optional

class SecurityToolBenchmarker:
    """Benchmark firewall throughput and password manager secret retrieval latency"""

    def __init__(self, firewall_host: str, password_manager_url: str, pm_api_key: str):
        self.firewall_host = firewall_host
        self.password_manager_url = password_manager_url.rstrip("/")
        self.pm_api_key = pm_api_key
        self.results: Dict = {
            "firewall_throughput_gbps": [],
            "password_manager_latency_ms": []
        }

    def benchmark_firewall_throughput(self, duration: int = 60, num_tests: int = 3) -> List[float]:
        """Benchmark firewall stateful throughput using iperf3 (https://github.com/esnet/iperf)

        Args:
            duration: Test duration per run in seconds
            num_tests: Number of test runs to average

        Returns:
            List of throughput values in Gbps
        """
        print(f"Starting firewall throughput benchmark: {num_tests} runs of {duration}s each")
        throughput_results = []

        for i in range(num_tests):
            print(f"Firewall test run {i+1}/{num_tests}")
            try:
                # Run iperf3 client against firewall host, TCP test, JSON output
                result = subprocess.run(
                    ["iperf3", "-c", self.firewall_host, "-t", str(duration), "-J"],
                    capture_output=True,
                    timeout=duration + 10
                )
                if result.returncode != 0:
                    print(f"iperf3 run {i+1} failed: {result.stderr.decode()}")
                    continue

                # Parse iperf3 JSON output
                iperf_data = json.loads(result.stdout.decode())
                # Get sum_received bits per second, convert to Gbps
                bps = iperf_data.get("end", {}).get("sum_received", {}).get("bits_per_second", 0)
                gbps = bps / 1e9
                throughput_results.append(gbps)
                print(f"Run {i+1} throughput: {gbps:.2f} Gbps")
                time.sleep(2)  # Cooldown between tests
            except Exception as e:
                print(f"Error in firewall benchmark run {i+1}: {str(e)}")

        self.results["firewall_throughput_gbps"] = throughput_results
        return throughput_results

    def benchmark_password_manager_latency(self, num_requests: int = 100) -> List[float]:
        """Benchmark password manager secret retrieval latency

        Args:
            num_requests: Number of secret retrieval requests to make

        Returns:
            List of latency values in milliseconds
        """
        print(f"Starting password manager latency benchmark: {num_requests} requests")
        latency_results = []
        headers = {"Authorization": f"Bearer {self.pm_api_key}"}

        for i in range(num_requests):
            try:
                start = time.time()
                # Retrieve a test secret (replace with valid item ID)
                response = requests.get(
                    f"{self.password_manager_url}/api/secrets/test_secret_123",
                    headers=headers,
                    timeout=5
                )
                response.raise_for_status()
                end = time.time()
                latency_ms = (end - start) * 1000
                latency_results.append(latency_ms)
                if (i+1) % 20 == 0:
                    print(f"Completed {i+1}/{num_requests} latency tests")
            except Exception as e:
                print(f"Error in latency test {i+1}: {str(e)}")

        self.results["password_manager_latency_ms"] = latency_results
        return latency_results

    def generate_report(self) -> Dict:
        """Generate benchmark report with statistics"""
        report = {}

        if self.results["firewall_throughput_gbps"]:
            throughput = self.results["firewall_throughput_gbps"]
            report["firewall_throughput"] = {
                "avg_gbps": round(statistics.mean(throughput), 2),
                "median_gbps": round(statistics.median(throughput), 2),
                "p99_gbps": round(statistics.quantiles(throughput, n=100)[98], 2) if len(throughput) >= 100 else None,
                "num_tests": len(throughput)
            }

        if self.results["password_manager_latency_ms"]:
            latency = self.results["password_manager_latency_ms"]
            report["password_manager_latency"] = {
                "avg_ms": round(statistics.mean(latency), 2),
                "median_ms": round(statistics.median(latency), 2),
                "p99_ms": round(statistics.quantiles(latency, n=100)[98], 2) if len(latency) >= 100 else None,
                "num_tests": len(latency)
            }

        return report

if __name__ == "__main__":
    # Benchmark configuration (replace with real values)
    benchmarker = SecurityToolBenchmarker(
        firewall_host="10.0.0.1",  # pfSense WAN IP
        password_manager_url="https://bitwarden.example.com",
        pm_api_key="your_bitwarden_api_key"
    )

    # Run benchmarks
    firewall_throughput = benchmarker.benchmark_firewall_throughput(duration=30, num_tests=5)
    pm_latency = benchmarker.benchmark_password_manager_latency(num_requests=200)

    # Generate and print report
    report = benchmarker.generate_report()
    print("\n=== Benchmark Report ===")
    print(json.dumps(report, indent=2))

    # Save report to file
    with open("security_benchmark_report.json", "w") as f:
        json.dump(report, f, indent=2)
    print("Report saved to security_benchmark_report.json")

iperf3 -c 10.0.0.1 -P 50000 -t 60 -J > firewall_benchmark.json

import { BitwardenClient } from "@bitwarden/sdk";
const client = new BitwardenClient({ apiKey: "your_api_key" });
const secret = await client.secrets.get("db_creds_456");

resource "aws_networkfirewall_rule_group" "allow_https" {
  name     = "allow-https"
  type     = "STATEFUL"
  capacity = 100
  rule_group {
    rules_source {
      stateful_rule {
        action = "PASS"
        header {
          protocol   = "TCP"
          source      = "10.0.0.0/8"
          destination = "0.0.0.0/0"
          destination_port = "443"
          direction = "ANY"
        }
      }
    }
  }
}