File size: 7,648 Bytes
7d944a4
d129093
 
 
 
 
 
7d944a4
09ab86d
 
 
 
7d944a4
 
 
 
d129093
 
 
 
09ab86d
 
 
d129093
 
 
 
 
 
 
 
 
 
 
 
 
c7eac83
09ab86d
d129093
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
09ab86d
 
 
 
 
 
 
 
 
 
 
 
7d944a4
 
 
 
 
 
 
 
09ab86d
 
 
 
 
 
 
 
 
 
 
 
 
 
7d944a4
 
 
 
 
09ab86d
7d944a4
08f99ec
09ab86d
08f99ec
7d944a4
 
 
 
09ab86d
7d944a4
09ab86d
d129093
 
09ab86d
7d944a4
 
 
08f99ec
09ab86d
08f99ec
 
 
 
7d944a4
09ab86d
7d944a4
 
 
 
 
 
 
08f99ec
7d944a4
08f99ec
 
 
 
 
 
 
09ab86d
7d944a4
08f99ec
7d944a4
 
 
09ab86d
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
d129093
 
 
09ab86d
7d944a4
 
09ab86d
 
 
 
 
 
 
 
 
 
7d944a4
 
09ab86d
7d944a4
 
 
08f99ec
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import requests
import os
from datetime import datetime, timedelta
from groq import Groq
from dotenv import load_dotenv
import logging
import asyncio
import aiohttp
from concurrent.futures import ThreadPoolExecutor
from functools import partial

# Configure logging
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)

# Load environment variables
load_dotenv()

# Configuration
GITHUB_TOKEN = os.getenv("GITHUB_TOKEN", "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")
GROQ_API_KEY = os.getenv("GROQ_API_KEY", "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")
REPOSITORIES = [
    "falcosecurity/rules",
    "SigmaHQ/sigma",
    "reversinglabs/reversinglabs-yara-rules",
    "elastic/detection-rules",
    "sublime-security/sublime-rules",
    "Yamato-Security/hayabusa-rules",
    "anvilogic-forge/armory",
    "chainguard-dev/osquery-defense-kit",
    "splunk/security_content",
    "Neo23x0/signature-base",
    "SlimKQL/Hunting-Queries-Detection-Rules"
]
DAYS_BACK = 1
TIMEOUT = 10  # seconds for each request

# GitHub API base URL
GITHUB_API_URL = "https://api.github.com"

# Groq client setup
groq_client = Groq(api_key=GROQ_API_KEY)

# FastAPI app
app = FastAPI(docs_url=None, redoc_url=None)

class RepositoryDetails(BaseModel):
    repo_name: str
    repo_url: str
    changes: str
    description: str
    context: str

async def fetch_with_aiohttp(url: str, headers: dict, params: dict, session: aiohttp.ClientSession) -> dict:
    """Asynchronous HTTP GET request with timeout."""
    try:
        async with session.get(url, headers=headers, params=params, timeout=aiohttp.ClientTimeout(total=TIMEOUT)) as response:
            response.raise_for_status()
            return await response.json()
    except Exception as e:
        logger.error(f"Error fetching {url}: {e}")
        return []

async def fetch_repository_changes(repo: str, days_back: int) -> list[str]:
    """Fetch recent commits and pull requests for a repository asynchronously."""
    try:
        logger.debug(f"Fetching changes for repository: {repo}")
        since_date = (datetime.now() - timedelta(days=days_back)).isoformat()
        headers = {
            "Authorization": f"token {GITHUB_TOKEN}",
            "Accept": "application/vnd.github.v3+json"
        }

        async with aiohttp.ClientSession() as session:
            # Fetch commits
            commits_url = f"{GITHUB_API_URL}/repos/{repo}/commits"
            commits_params = {"since": since_date, "per_page": 100}  # Limit to 100 for efficiency
            logger.debug(f"Fetching commits from: {commits_url}")
            commits = await fetch_with_aiohttp(commits_url, headers, commits_params, session)
            logger.debug(f"Found {len(commits)} commits for {repo}")

            # Fetch pull requests
            prs_url = f"{GITHUB_API_URL}/repos/{repo}/pulls"
            prs_params = {"state": "all", "sort": "updated", "direction": "desc", "per_page": 100}
            logger.debug(f"Fetching pull requests from: {prs_url}")
            prs = await fetch_with_aiohttp(prs_url, headers, prs_params, session)
            logger.debug(f"Found {len(prs)} pull requests for {repo}")

        # Extract changes
        changes = []
        for commit in commits:
            changes.append(f"Commit: {commit['commit']['message']}")
        cutoff_date = datetime.now() - timedelta(days=days_back)
        for pr in prs:
            updated_at = datetime.strptime(pr["updated_at"], "%Y-%m-%dT%H:%M:%SZ")
            if updated_at >= cutoff_date:
                changes.append(f"PR: {pr['title']} - {pr['body'] or 'No description'}")

        logger.debug(f"Total changes for {repo}: {len(changes)}")
        return changes

    except Exception as e:
        logger.error(f"Error fetching changes for {repo}: {e}")
        return []

def summarize_changes_with_deepseek(repo: str, changes: list[str]) -> dict:
    """Use Groq's DeepSeek model to summarize changes (synchronous due to Groq SDK limitations)."""
    try:
        logger.debug(f"Summarizing changes for repository: {repo}")
        prompt = f"""
        Analyze the following changes made to detection rules in the GitHub repository {repo}:
        {', '.join(changes[:50])}  # Limit to 50 changes to avoid token overflow
        
        Provide a detailed response with two sections:
        - Description: Summarize what changes were made.
        - Context: Explain why these changes might be required.
        """
        logger.debug(f"Sending prompt to DeepSeek: {prompt[:100]}...")
        response = groq_client.chat.completions.create(
            model="deepseek-chat",
            messages=[{"role": "user", "content": prompt}],
            max_tokens=500,
            temperature=0.7
        )
        summary = response.choices[0].message.content
        logger.debug(f"Received summary from DeepSeek: {summary[:100]}...")

        # Extract description and context with fallback
        description = "Description not found."
        context = "Context not found."
        if "Description:" in summary and "Context:" in summary:
            description = summary.split("Description:")[1].split("Context:")[0].strip()
            context = summary.split("Context:")[1].strip()
        else:
            description = summary

        return {"description": description, "context": context}

    except Exception as e:
        logger.error(f"Error summarizing changes for {repo}: {e}")
        return {"description": "Error occurred.", "context": "Error occurred."}

async def process_repository(repo: str, days_back: int) -> RepositoryDetails:
    """Process a single repository and return its details."""
    changes = await fetch_repository_changes(repo, days_back)
    if changes:
        summary = await asyncio.get_event_loop().run_in_executor(
            None, partial(summarize_changes_with_deepseek, repo, changes)
        )
        return RepositoryDetails(
            repo_name=f"{repo} (+{len(changes)})",
            repo_url=f"https://github.com/{repo}",
            changes="\n".join(changes),
            description=summary["description"],
            context=summary["context"]
        )
    else:
        return RepositoryDetails(
            repo_name=f"{repo} (No changes)",
            repo_url=f"https://github.com/{repo}",
            changes=f"No changes detected in the last {days_back} day(s).",
            description="No changes detected.",
            context="No context available."
        )

@app.get("/monitor", response_model=list[RepositoryDetails])
async def monitor_repositories():
    """Single API endpoint to fetch and summarize changes for all repositories."""
    try:
        logger.debug("Starting to monitor repositories")
        tasks = [process_repository(repo, DAYS_BACK) for repo in REPOSITORIES]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        
        # Handle any exceptions in results
        final_results = []
        for result in results:
            if isinstance(result, Exception):
                logger.error(f"Error processing a repository: {result}")
                continue
            final_results.append(result)

        logger.debug("Finished monitoring repositories")
        return final_results

    except Exception as e:
        logger.error(f"Error in monitor_repositories: {e}")
        raise HTTPException(status_code=500, detail=str(e))

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)