minhan6559's picture
Upload app.py
31bcda8 verified
#!/usr/bin/env python3
"""
Streamlit Web App for Cybersecurity Agent Pipeline
A simple web interface for uploading log files and running the cybersecurity analysis pipeline
with different LLM models.
"""
import os
import sys
import tempfile
import shutil
import time
import streamlit as st
from pathlib import Path
from typing import Dict, Any, Optional
# Add project root to path for agent imports
project_root = Path(__file__).parent
sys.path.insert(0, str(project_root))
from src.full_pipeline.simple_pipeline import analyze_log_file
from dotenv import load_dotenv
from huggingface_hub import login as huggingface_login
from huggingface_hub.utils import HfHubHTTPError
load_dotenv()
def get_model_providers() -> Dict[str, Dict[str, str]]:
"""Get available model providers and their models."""
return {
"Google GenAI": {
"gemini-2.0-flash": "google_genai:gemini-2.0-flash",
"gemini-2.0-flash-lite": "google_genai:gemini-2.0-flash-lite",
"gemini-2.5-flash-lite": "google_genai:gemini-2.5-flash-lite",
},
"Groq": {
"openai/gpt-oss-120b": "groq:openai/gpt-oss-120b",
"moonshotai/kimi-k2-instruct-0905": "groq:moonshotai/kimi-k2-instruct-0905",
},
"OpenAI": {
"gpt-5-mini": "openai:gpt-5-mini",
"gpt-5": "openai:gpt-5",
"gpt-4.1-mini": "openai:gpt-4.1-mini",
},
}
def get_api_key_help() -> Dict[str, str]:
"""Get API key help information for each provider."""
return {
"Google GenAI": "https://aistudio.google.com/app/apikey",
"Groq": "https://console.groq.com/keys",
"OpenAI": "https://platform.openai.com/api-keys",
}
def setup_temp_directories(temp_dir: str) -> Dict[str, str]:
"""Setup temporary directories for the pipeline."""
log_files_dir = os.path.join(temp_dir, "log_files")
analysis_dir = os.path.join(temp_dir, "analysis")
final_response_dir = os.path.join(temp_dir, "final_response")
os.makedirs(log_files_dir, exist_ok=True)
os.makedirs(analysis_dir, exist_ok=True)
os.makedirs(final_response_dir, exist_ok=True)
return {
"log_files": log_files_dir,
"analysis": analysis_dir,
"final_response": final_response_dir,
}
def save_uploaded_file(uploaded_file, temp_dir: str) -> str:
"""Save uploaded file to temporary directory."""
log_files_dir = os.path.join(temp_dir, "log_files")
file_path = os.path.join(log_files_dir, uploaded_file.name)
with open(file_path, "wb") as f:
f.write(uploaded_file.getbuffer())
return file_path
def run_analysis(
log_file_path: str,
model_name: str,
query: str,
temp_dirs: Dict[str, str],
api_key: str,
provider: str,
max_log_analysis_iterations: int,
max_retrieval_iterations: int,
progress_callback=None,
) -> Dict[str, Any]:
"""Run the cybersecurity analysis pipeline."""
# Set environment variable for API key
if provider == "Google GenAI":
os.environ["GOOGLE_API_KEY"] = api_key
elif provider == "Groq":
os.environ["GROQ_API_KEY"] = api_key
elif provider == "OpenAI":
os.environ["OPENAI_API_KEY"] = api_key
try:
# Run the analysis pipeline
result = analyze_log_file(
log_file=log_file_path,
query=query,
tactic=None,
model_name=model_name,
temperature=0.1,
max_log_analysis_iterations=max_log_analysis_iterations,
max_retrieval_iterations=max_retrieval_iterations,
log_agent_output_dir=temp_dirs["analysis"],
response_agent_output_dir=temp_dirs["final_response"],
progress_callback=progress_callback,
)
return {"success": True, "result": result}
except Exception as e:
return {"success": False, "error": str(e)}
@st.cache_resource
def initialize_hf_login():
"""Initialize Hugging Face login only once."""
hf_token = os.getenv("HF_TOKEN")
if hf_token:
try:
# Check if already logged in by trying to get user info
from huggingface_hub import whoami
whoami()
return True
except (HfHubHTTPError, Exception):
# Not logged in, try to login
try:
huggingface_login(token=hf_token)
return True
except Exception as e:
st.warning(f"Failed to login to Hugging Face: {e}")
return False
return False
def main():
"""Main Streamlit app."""
# Initialize HF login (cached)
initialize_hf_login()
st.set_page_config(
page_title="Cybersecurity Agent Pipeline", page_icon="🛡️", layout="wide"
)
st.title("Cybersecurity Agent Pipeline")
st.markdown(
"Upload a log file and analyze it using advanced LLM-based cybersecurity agents."
)
# Sidebar for configuration
with st.sidebar:
st.header("Configuration")
# Model selection
providers = get_model_providers()
selected_provider = st.selectbox(
"Select Model Provider", list(providers.keys())
)
available_models = providers[selected_provider]
selected_model_display = st.selectbox(
"Select Model", list(available_models.keys())
)
selected_model = available_models[selected_model_display]
# API Key input with help
st.subheader("API Key")
api_key_help = get_api_key_help()
with st.expander("How to get API key", expanded=False):
st.markdown(f"**{selected_provider}**:")
st.markdown(f"[Get API Key]({api_key_help[selected_provider]})")
api_key = st.text_input(
f"Enter {selected_provider} API Key",
type="password",
help=f"Your {selected_provider} API key",
)
# Main content area
col1, col2 = st.columns([2, 1])
with col1:
st.header("Upload Log File")
uploaded_file = st.file_uploader(
"Choose a JSON log file",
type=["json"],
help="Upload a JSON log file from the Mordor dataset or similar security logs",
)
with col2:
st.header("Analysis Status")
if uploaded_file is not None:
st.success(f"File uploaded: {uploaded_file.name}")
st.info(f"Size: {uploaded_file.size:,} bytes")
else:
st.warning("Please upload a log file")
# Run analysis button
if st.button(
"Run Analysis", type="primary", disabled=not (uploaded_file and api_key)
):
if not uploaded_file:
st.error("Please upload a log file first.")
return
if not api_key:
st.error("Please enter your API key.")
return
# Create temporary directory
temp_dir = tempfile.mkdtemp(prefix="cyber_agent_")
try:
# Setup directories
temp_dirs = setup_temp_directories(temp_dir)
# Save uploaded file
log_file_path = save_uploaded_file(uploaded_file, temp_dir)
# Show progress
progress_bar = st.progress(0)
status_text = st.empty()
status_text.text("Initializing analysis...")
progress_bar.progress(10)
# Start timing
start_time = time.time()
# Create progress callback
def update_progress(progress: int, message: str):
progress_bar.progress(progress)
status_text.text(message)
# Run analysis
analysis_result = run_analysis(
log_file_path=log_file_path,
model_name=selected_model,
query="",
temp_dirs=temp_dirs,
api_key=api_key,
provider=selected_provider,
max_log_analysis_iterations=2,
max_retrieval_iterations=2,
progress_callback=update_progress,
)
# Calculate execution time
end_time = time.time()
execution_time = end_time - start_time
progress_bar.progress(90)
status_text.text("Finalizing results...")
if analysis_result["success"]:
progress_bar.progress(100)
status_text.text("Analysis completed successfully!")
# Display results
st.header("Analysis Results")
result = analysis_result["result"]
# Show key metrics
col1, col2, col3 = st.columns(3)
with col1:
assessment = result.get("log_analysis_result", {}).get(
"overall_assessment", "Unknown"
)
st.metric("Overall Assessment", assessment)
with col2:
abnormal_events = result.get("log_analysis_result", {}).get(
"abnormal_events", []
)
st.metric("Abnormal Events", len(abnormal_events))
with col3:
st.metric("Execution Time", f"{execution_time:.2f}s")
# Show markdown report
markdown_report = result.get("markdown_report", "")
if markdown_report:
st.header("Detailed Report")
st.markdown(markdown_report)
else:
st.warning("No detailed report generated.")
else:
st.error(f"Analysis failed: {analysis_result['error']}")
st.exception(analysis_result["error"])
finally:
# Cleanup temporary directory
try:
shutil.rmtree(temp_dir)
except Exception as e:
st.warning(f"Could not clean up temporary directory: {e}")
# Footer
st.markdown("---")
st.markdown(
"**Cybersecurity Agent Pipeline** - Powered by LangGraph and LangChain | "
"Built for educational purposes demonstrating LLM-based multi-agent systems"
)
if __name__ == "__main__":
main()