git clone https://github.com/icarax/fortibled-credential-watchdog.git
cd fortibled-credential-watchdog

# Create a virtual environment (recommended)
python -m venv .venv
source .venv/bin/activate   # on Windows: .venv\Scripts\activate

# Install dependencies
pip install --upgrade pip
pip install requests python-dotenv tqdm

# Initialise a new npm project (if you don't have one)
npm init -y

# Install runtime deps
npm install axios dotenv

# Install TypeScript & typings (if you want TS)
npm install --save-dev typescript @types/node ts-node
npx tsc --init   # creates a basic tsconfig.json

# .env  (do NOT commit this file!)
HIBP_API_KEY=your_haveibeenpwned_api_key_here
OTX_API_KEY=your_alienvault_otx_api_key_here   # optional

#!/usr/bin/env python3
"""
FortiBleed Credential Watchdog – Python version
------------------------------------------------
- Reads username:password lines from a file (default: creds.txt)
- Checks each password against HaveIBeenPwned (k‑anonymity)
- Pulls FortiBleed‑tagged IPv4 IOCs from AlienVault OTX
- Prints compromised accounts and any IOC IPs that appear in a supplied log file
"""

import os
import sys
import hashlib
import time
from pathlib import Path
from typing import List, Tuple, Set

import requests
from dotenv import load_dotenv
from tqdm import tqdm  # nice progress bar

# ----------------------------------------------------------------------
# Configuration &nbsp
load_dotenv()  # pulls variables from .env into os.environ
HIBP_API_KEY = os.getenv("HIBP_API_KEY")
OTX_API_KEY = os.getenv("OTX_API_KEY")  # optional

if not HIBP_API_KEY:
    sys.exit("❌ HIBP_API_KEY not set. Get a free key at https://haveibeenpwned.com/API/v3")

HIBP_URL = "https://api.pwnedpasswords.com/range/{prefix}"
OTX_URL = "https://otx.alienvault.com/api/v1/pulses/subscribed"

HEADERS = {
    "User-Agent": "FortiBleed-Watchdog/1.0 (+https://icarax.blog)",
    "hibp-api-key": HIBP_API_KEY,
}
if OTX_API_KEY:
    HEADERS["X-OTX-API-KEY"] = OTX_API_KEY

# ----------------------------------------------------------------------
def sha1_hash(text: str) -> str:
    """Return uppercase SHA‑1 hex digest."""
    return hashlib.sha1(text.encode("utf-8")).hexdigest().upper()


def hibp_lookup(hash_suffix: str) -> Set[str]:
    """
    Query HIBP for hash suffixes.
    Returns a set of matching suffixes (the part after the first 5 chars).
    """
    url = HIBP_URL.format(prefix=hash_suffix)
    try:
        resp = requests.get(url, headers=HEADERS, timeout=10)
        resp.raise_for_status()
    except requests.RequestException as exc:
        print(f"⚠️  HIBP request failed for {hash_suffix}: {exc}")
        return set()

    # HIBP returns lines like: 310A5F7F8E9C...:4
    lines = resp.text.splitlines()
    return {line.split(":")[0] for line in lines if ":" in line}


def fetch_otx_iocs() -> Set[str]:
    """
    Pull all IPv4 IOCs from pulses tagged with 'FortiBleed'.
    Returns a set of IP strings.
    """
    if not OTX_API_KEY:
        return set()

    ips: Set[str] = set()
    try:
        resp = requests.get(
            OTX_URL,
            headers=HEADERS,
            params={"limit": 50},  # adjust as needed
            timeout=15,
        )
        resp.raise_for_status()
        data = resp.json()
        for pulse in data.get("results", []):
            # Only consider pulses that have the FortiBleed tag
            if any(t.lower() == "fortibled" for t in pulse.get("tags", [])):
                for indicator in pulse.get("indicators", []):
                    if indicator["type"] == "IPv4":
                        ips.add(indicator["indicator"])
    except Exception as exc:
        print(f"⚠️  OTX request failed: {exc}")
    return ips


def process_credentials(file_path: Path) -> List[Tuple[str, str, bool]]:
    """
    Returns list of (username, password, compromised_flag).
    """
    results: List[Tuple[str, str, bool]] = []
    with file_path.open() as f:
        for line_num, raw in enumerate(f, 1):
            line = raw.strip()
            if not line or line.startswith("#"):
                continue
            if ":" not in line:
                print(f"⚠️  Skipping malformed line {line_num}: {line}")
                continue
            username, password = line.split(":", 1)
            pwd_hash = sha1_hash(password)
            prefix, suffix = pwd_hash[:5], pwd_hash[5:]

            matches = hibp_lookup(suffix)
            compromised = bool(matches)
            results.append((username, password, compromised))

            # Be nice to HIBP – they allow 1 request per 1.5 seconds on free tier
            time.sleep(1.2)
    return results


def main():
    import argparse

    parser = argparse.ArgumentParser(
        description="Check credentials against HIBP and flag FortiBleed IOCs"
    )
    parser.add_argument(
        "-f",
        "--file",
        type=Path,
        default=Path("creds.txt"),
        help="Path to username:password file",
    )
    parser.add_argument(
        "-l",
        "--log",
        type=Path,
        help="Optional log file to scan for FortiBleed IOC IPs",
    )
    args = parser.parse_args()

    if not args.file.is_file():
        sys.exit(f"❌ Credentials file not found: {args.file}")

    print("🔎 Loading FortiBleed IOCs from OTX …")
    fortibled_ips = fetch_otx_iocs()
    print(f"   Found {len(fortibled_ips)} IPv4 IOCs tagged FortiBleed")

    print(f"🔎 Scanning credentials in {args.file} …")
    cred_results = process_credentials(args.file)

    compromised = [(u, p) for u, p, c in cred_results if c]
    print("\n=== COMPROMISED CREDENTIALS ===")
    if compromised:
        for user, pwd in compromised:
            # Never log the plain password in prod – just show it's compromised
            print(f"🔴 {user}:<REDACTED>")
    else:
        print("✅ No credentials found in HIBP breach corpus.")

    # Optional: scan a log for any of the FortiBleed IPs
    if args.log and args.log.is_file():
        print(f"\n🔎 Scanning log file {args.log} for FortiBleed IPs …")
        log_text = args.log.read_text()
        found_ips = [ip for ip in fortibled_ips if ip in log_text]
        if found_ips:
            print("🚩 FortiBleed IPs observed in log:")
            for ip in found_ips:
                print(f"   - {ip}")
        else:
            print("✅ No FortiBleed IOC IPs seen in the supplied log.")
    elif args.log:
        print(f"⚠️  Log file not found: {args.log}")

if __name__ == "__main__":
    main()

# 1️⃣ Prepare a test file (username:password per line)
echo "alice:S3cr3t!" > creds.txt
echo "bob:Password123" >> creds.txt

# 2️⃣ Execute
python watchdog.py -f creds.txt   # add -l /var/log/auth.log if you have a log to scan

#!/usr/bin/env node
/**
 * FortiBleed Credential Watchdog – TypeScript version
 * --------------------------------------------------
 * Same functionality as the Python script, but runs on Node.js.
 * Requires Node >=14 and the packages installed in Step 2.
 */

import * as dotenv from "dotenv";
import * as fs from "fs";
import * as path from "path";
import axios from "axios";
import { createHash } from "crypto";

// Load .env
dotenv.config();

const HIBP_API_KEY = process.env.HIBP_API_KEY;
const OTX_API_KEY = process.env.OTX_API_KEY; // optional

if (!HIBP_API_KEY) {
  console.error("❌ HIBP_API_KEY is missing. Get a free key at https://haveibeenpwned.com/API/v3");
  process.exit(1);
}

const HIBP_URL = (prefix: string) => `https://api.pwnedpasswords.com/range/${prefix}`;
const OTX_URL = "https://otx.alienvault.com/api/v1/pulses/subscribed";

const axiosInstance = axios.create({
  timeout: 10000,
  headers: {
    "User-Agent": "FortiBleed-Watchdog/1.0 (+https://icarax.blog)",
    "hibp-api-key": HIBP_API_KEY,
    ...(OTX_API_KEY ? { "X-OTX-API-KEY": OTX_API_KEY } : {}),
  },
});

// ---------------------------------------------------------------------
// Helper functions
// ---------------------------------------------------------------------
function sha1(str: string): string {
  return createHash("sha1").update(str, "utf8").digest("hex").toUpperCase();
}

/**
 * Query HIBP for a hash suffix (everything after the first 5 chars).
 * Returns a Set of matching suffixes.
 */
async function hibpLookup(suffix: string): Promise<Set<string>> {
  try {
    const { data } = await axiosInstance.get(HIBP_URL(suffix.slice(0, 5)));
    const lines = data.split("\n");
    const matches = new Set<string>();
    for (const line of lines) {
      if (line.includes(":")) {
        const [hashPart] = line.split(":");
        if (hashPart.startsWith(suffix)) matches.add(hashPart);
      }
    }
    return matches;
  } catch (err: any) {
    console.warn(`⚠️  HIBP lookup failed for ${suffix}: ${err.message}`);
    return new Set();
  }
}

/**
 * Pull IPv4 IOCs from OTX pulses tagged with 'FortiBleed'.
 */
async function fetchOTXIocs(): Promise<Set<string>> {
  if (!OTX_API_KEY) return new Set();
  try {
    const { data } = await axiosInstance.get(OTX_URL, {
      params: { limit: 50 },
    });
    const ips = new Set<string>();
    for (const pulse of data.results ?? []) {
      const hasTag = pulse.tags?.some((t: string) => t.toLowerCase() === "fortibled");
      if (!hasTag) continue;
      for (const ind of pulse.indicators ?? []) {
        if (ind.type === "IPv4") ips.add(ind.indicator);
      }
    }
    return ips;
  } catch (err: any) {
    console.warn(`⚠️  OTX request failed: ${err.message}`);
    return new Set();
  }
}

/**
 * Read credentials file (username:password per line).
 */
function readCredentials(filePath: string): Array<{ user: string; pass: string }> {
  const content = fs.readFileSync(filePath, "utf8");
  return content
    .split("\n")
    .map((l) => l.trim())
    .filter((l) => l && !l.startsWith("#"))
    .map((l) => {
      const [user, pass] = l.split(":");
      if (!user || !pass) throw new Error(`Malformed line: ${l}`);
      return { user, pass };
    });
}

/**
 * Main routine
 */
async function main() {
  const args = process.argv.slice(2);
  const credFile = args.includes("-f") || args.includes("--file")
    ? args[args.indexOf("-f") !== -1 ? args.indexOf("-f") : args.indexOf("--file") + 1]
    : "creds.txt";
  const logFile = args.includes("-l") || args.includes("--log")
    ? args[args.indexOf("-l") !== -1 ? args.indexOf("-l") : args.indexOf("--log") + 1]
    : undefined;

  if (!fs.existsSync(credFile)) {
    console.error(`❌ Credentials file not found: ${credFile}`);
    process.exit(1);
  }

  console.log("🔎 Fetching FortiBleed IOCs from OTX …");
  const fortibledIPs = await fetchOTXIocs();
  console.log(`   Found ${fortibledIPs.size} IPv4 IOCs tagged FortiBleed`);

  console.log(`🔎 Scanning credentials in ${credFile} …`);
  const creds = readCredentials(credFile);
  const compromised: Array<{ user: string; pass: string }> = [];

  for (const { user, pass } of creds) {
    const hash = sha1(pass);
    const prefix = hash.slice(0, 5);
    const suffix = hash.slice(5);
    const matches = await hibpLookup(suffix);
    if (matches.size > 0) compromised.push({ user, pass });

    // Respect HIBP rate limit (free tier: 1 request per 1.5s)
    await new Promise((r) => setTimeout(r, 1500));
  }

  console.log("\n=== COMPROMISED CREDENTIALS ===");
  if (compromised.length) {
    for const { user } of compromised) {
      console.log(`🔴 ${user}:<REDACTED>`);
    }
  } else {
    console.log("✅ No credentials appear in HIBP breach corpus.");
  }

  // Optional log scan
  if (logFile && fs.existsSync(logFile)) {
    console.log(`\n🔎 Scanning log ${logFile} for FortiBleed IPs …`);
    const logText = fs.readFileSync(logFile, "utf8");
    const found = [...fortibledIPs].filter((ip) => logText.includes(ip));
    if (found.length) {
      console.log("🚩 FortiBleed IPs observed in log:");
      for (const ip of found) console.log(`   - ${ip}`);
    } else {
      console.log("✅ No FortiBleed IOC IPs seen in the supplied log.");
    }
  } else if (logFile) {
    console.warn(`⚠️  Log file not found: ${logFile}`);
  }
}

main().catch((err) => {
  console.error("❌ Unexpected error:", err);
  process.exit(1);
});

# Compile & run
npx ts-node watchdog.ts -f creds.txt   # add -l /var/log/auth.log if desired

npx tsc   # produces watchdog.js
node watchdog.js -f creds.txt

# .gitignore
.env
*.log

# Python
time.sleep(1.2)   # HIBP free tier: ≤1 request per 1.5 seconds

// TypeScript
await new Promise(r => setTimeout(r, 1500));

import backoff  # pip install backoff

@backoff.on_exception(backoff.expo,
                      (requests.exceptions.RequestException,),
                      max_tries=5)
def hibp_lookup_with_backoff(suffix):
    ...

// Simple manual back‑off
async function hibpLookupWithBackoff(suffix: string, attempt = 0): Promise<Set<string>> {
  try {
    return await hibpLookup(suffix);
  } catch (e) {
    if (attempt < 5) {
      await new Promise(r => setTimeout(r, Math.pow(2, attempt) * 500));
      return hibpLookupWithBackoff(suffix, attempt + 1);
    }
    throw e;
  }
}

from functools import lru_cache

@lru_cache(maxsize=256_000)
def cached_hibp_lookup(suffix):
    return hibp_lookup(suffix)

const cache = new Map<string, Set<string>>();
async function cachedHibpLookup(suffix: string): Promise<Set<string>> {
  if (cache.has(suffix)) return cache.get(suffix)!;
  const res = await hibpLookup(suffix);
  cache.set(suffix, res);
  return res;
}