Python Intermediate Project – Cyber Analyst Academy

The modern cybersecurity landscape is highly dynamic, with adversaries constantly evolving their attack techniques. Organizations must proactively monitor, detect, and respond to cyber threats before they escalate into full-scale breaches.

This capstone project aims to develop a Threat Intelligence Automation & Anomaly Detection System that integrates functional programming, concurrency, databases, APIs, data science, and testing to collect, process, analyze, and visualize threat intelligence. The system will scan dark web sources, threat intelligence feeds, and network logs to identify potential indicators of compromise (IOCs).

By completing this project, you will:

Apply functional programming & concurrency to process threat data efficiently.
Utilize databases for storing and querying threat intelligence records.
Build secure web APIs for real-time threat intelligence sharing.
Implement machine learning models for anomaly detection in cybersecurity.
Conduct robust testing & debugging to ensure system reliability.

Python Functional & Concurrent Programming – Efficient Threat Data Collection

We use functional programming and concurrency to efficiently retrieve multiple threat feeds.

Code: Scraping Threat Intelligence Feeds

import requests
import concurrent.futures

# List of threat intelligence sources
THREAT_FEEDS = {
    "alienvault": "https://otx.alienvault.com/api/v1/indicators",
    "virustotal": "https://www.virustotal.com/api/v3/intelligence/hunting_rulesets"
}

def fetch_data(source, url):
    headers = {
        "User-Agent": "ThreatIntelBot",
        "Authorization": "Bearer YOUR_API_KEY"  # Replace with valid API Key
    }
    response = requests.get(url, headers=headers)
    return {source: response.json()} if response.status_code == 200 else {source: None}

# Concurrently fetch threat intelligence from multiple sources
def collect_threat_intel():
    with concurrent.futures.ThreadPoolExecutor() as executor:
        results = executor.map(lambda source: fetch_data(source, THREAT_FEEDS[source]), THREAT_FEEDS)
    return dict(results)

# Run the scraper
threat_data = collect_threat_intel()
print(threat_data)

🔹 Concepts Used: Functional programming (map()), ThreadPoolExecutor for multithreading.
🔹 Purpose: Speed up threat intelligence collection from multiple sources simultaneously.

Python Database Interactions – Storing and Querying Threat Data

A PostgreSQL database will be used to store threat indicators for further analysis.

Code: Database Schema & Ingestion Script

import sqlite3

# Connect to SQLite (Use PostgreSQL in production)
conn = sqlite3.connect('threat_intelligence.db')
cursor = conn.cursor()

# Create table
cursor.execute('''
CREATE TABLE IF NOT EXISTS threats (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    source TEXT,
    type TEXT,
    value TEXT UNIQUE,
    timestamp DATETIME DEFAULT CURRENT_TIMESTAMP
)
''')

def insert_threat(source, type, value):
    try:
        cursor.execute("INSERT INTO threats (source, type, value) VALUES (?, ?, ?)", (source, type, value))
        conn.commit()
    except sqlite3.IntegrityError:
        print(f"Duplicate threat detected: {value}")

# Example insertion
insert_threat("AlienVault", "IP", "192.168.1.100")

# Query stored threats
cursor.execute("SELECT * FROM threats")
print(cursor.fetchall())

conn.close()

🔹 Concepts Used: SQLite/PostgreSQL, SQL Transactions, Unique Constraints.
🔹 Purpose: Efficiently store threat indicators and prevent duplicates.

Python Web & API Development – Exposing Threat Intelligence via API

We build a Flask REST API to expose stored threat intelligence data.

Code: Flask API for Querying Threat Data

from flask import Flask, jsonify, request
import sqlite3

app = Flask(__name__)

def query_db(query, args=(), one=False):
    conn = sqlite3.connect('threat_intelligence.db')
    cursor = conn.cursor()
    cursor.execute(query, args)
    result = cursor.fetchall()
    conn.close()
    return (result[0] if result else None) if one else result

@app.route('/threats', methods=['GET'])
def get_threats():
    threats = query_db("SELECT * FROM threats")
    return jsonify(threats)

if __name__ == '__main__':
    app.run(debug=True)

🔹 Concepts Used: Flask, REST API, Database Queries.
🔹 Purpose: Expose threat intelligence data for real-time monitoring.

Python Data Science & Machine Learning – Detecting Cyber Anomalies

We train a machine learning model to detect anomalies in network logs.

Code: ML-Based Anomaly Detection Using Scikit-Learn

import pandas as pd
from sklearn.ensemble import IsolationForest

# Load sample network log data
data = pd.read_csv("network_logs.csv")
features = data[['bytes_sent', 'bytes_received', 'response_time']]

# Train Isolation Forest
model = IsolationForest(contamination=0.05)
data['anomaly'] = model.fit_predict(features)

# Display anomalies
anomalies = data[data['anomaly'] == -1]
print("Detected Anomalies:\n", anomalies)

🔹 Concepts Used: Isolation Forest for detecting outliers in network logs.
🔹 Purpose: Identify suspicious network traffic indicating malicious activity.

Python Testing & Debugging – Ensuring Robust Code

We use pytest for API testing & debugging.

Code: Unit Tests for Threat API

import pytest
import requests

API_URL = "http://127.0.0.1:5000/threats"

def test_threat_api():
    response = requests.get(API_URL)
    assert response.status_code == 200
    assert isinstance(response.json(), list)

if __name__ == "__main__":
    pytest.main()

🔹 Concepts Used: Pytest, API Testing.
🔹 Purpose: Ensure the API is functional and returns valid data.

This project simulates real-world threat intelligence automation, where timely detection of threats can prevent catastrophic cyberattacks.

Adversaries constantly evolve—organizations must adopt AI-driven threat hunting to stay ahead.
Functional programming & concurrency accelerate threat detection, preventing zero-day exploits.
Machine learning enhances detection capabilities, helping identify anomalous behaviors.

Cyber warfare is an arms race. This project arms you with Python-driven cybersecurity intelligence—a crucial weapon in defending digital infrastructure.