Backend satyasai

.gitignore

@@ -0,0 +1,6 @@
+venv/
+__pycache__/
+*.pyc
+.env
+backend/uploads/
+*.zip

backend/.gitignore

@@ -0,0 +1,19 @@
+# Python cache
+__pycache__/
+*.py[cod]
+
+# Virtual environment
+venv/
+
+# Environment variables
+.env
+
+# Uploaded files
+uploads/*
+
+# Keep the uploads folder
+!uploads/.gitkeep
+
+# Cybersecurity cache
+../cybersecurity/__pycache__/
+../cybersecurity/**/__pycache__/

backend/app/api/upload.py

@@ -0,0 +1,16 @@
+from fastapi import APIRouter, UploadFile, File
+
+from app.services.upload_service import UploadService
+from app.schemas.upload import UploadResponse
+
+router = APIRouter(
+    prefix="/upload",
+    tags=["File Upload"]
+)
+
+service = UploadService()
+
+
+@router.post("/", response_model=UploadResponse)
+def upload_file(file: UploadFile = File(...)):
+    return service.upload_file(file)

backend/app/main.py

@@ -1,13 +1,20 @@
+import sys
+import os
+sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "../../")))
+
 from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
 
+from app.api.upload import router as upload_router
 from app.api.auth import router as auth_router
 from app.core.config import settings
 from app.db.session import Base, engine
 
-# Automatically initialize database tables on startup
+
+# Automatically initialize database tables
 Base.metadata.create_all(bind=engine)
 
+
 app = FastAPI(
     title=settings.PROJECT_NAME,
     description="AI-Powered Malware Classification & Threat Detection System Backend",
@@ -16,24 +23,26 @@
     redoc_url="/redoc"
 )
 
-# Configure Cross-Origin Resource Sharing (CORS) for Frontend React integration
+
+# Configure CORS for frontend React integration
 app.add_middleware(
     CORSMiddleware,
-    allow_origins=["*"],  # Restrict in production environment
+    allow_origins=["*"],
     allow_credentials=True,
     allow_methods=["*"],
     allow_headers=["*"],
 )
 
-# Register API Routers
+
+# Register API routers
+app.include_router(upload_router)
 app.include_router(auth_router, prefix=settings.API_V1_STR)
 
 
 @app.get("/", tags=["Health Check"])
 def root():
-    """Health check endpoint to verify backend service status."""
     return {
         "status": "healthy",
         "service": settings.PROJECT_NAME,
         "docs": "/docs"
-    }
+    }

backend/app/schemas/upload.py

@@ -0,0 +1,68 @@
+from datetime import datetime
+
+from pydantic import BaseModel
+
+
+class FileInformation(BaseModel):
+    filename: str
+    extension: str
+    file_type: str
+    size: int
+
+
+class ValidationResult(BaseModel):
+    valid: bool
+    extension: str
+    size: int
+
+
+class MetadataResult(BaseModel):
+    filename: str
+    extension: str
+    file_type: str
+    size: int
+    created_time: str
+    modified_time: str
+
+
+class HashResult(BaseModel):
+    md5: str
+    sha256: str
+
+
+class PESection(BaseModel):
+    name: str
+    virtual_address: str
+    virtual_size: int
+    raw_size: int
+    characteristics: str
+
+
+class PEAnalysis(BaseModel):
+    headers: dict
+    sections: list[PESection]
+    suspicious_characteristics: list[str]
+
+
+class ImportsResult(BaseModel):
+    dlls: list[str]
+    apis: list[str]
+
+
+class StaticAnalysisResult(BaseModel):
+    file: FileInformation
+    validation: ValidationResult
+    metadata: MetadataResult
+    hashes: HashResult
+    pe_analysis: PEAnalysis
+    imports: ImportsResult
+    strings: list[str]
+
+
+class UploadResponse(BaseModel):
+    message: str
+    original_filename: str
+    stored_filename: str
+    file_size: int
+    upload_time: datetime
+    analysis: StaticAnalysisResult

backend/app/services/upload_service.py

@@ -0,0 +1,68 @@
+import os
+import uuid
+from datetime import datetime
+from fastapi import UploadFile, HTTPException
+from cybersecurity.file_analysis_pipeline import analyze_file
+
+UPLOAD_FOLDER = "uploads"
+
+os.makedirs(UPLOAD_FOLDER, exist_ok=True)
+
+ALLOWED_EXTENSIONS = {".exe", ".dll", ".zip", ".pdf", ".doc", ".docx"}
+
+MAX_FILE_SIZE = 100 * 1024 * 1024  # 100 MB
+
+
+class UploadService:
+
+    def validate_file(self, file: UploadFile):
+
+        extension = os.path.splitext(file.filename)[1].lower()
+
+        if extension not in ALLOWED_EXTENSIONS:
+            raise HTTPException(status_code=400, detail="Unsupported file type")
+
+    def save_file(self, file: UploadFile):
+
+        unique_filename = f"{uuid.uuid4()}_{file.filename}"
+
+        file_path = os.path.join(UPLOAD_FOLDER, unique_filename)
+
+        try:
+            with open(file_path, "wb") as buffer:
+                content = file.file.read()
+
+                if len(content) > MAX_FILE_SIZE:
+                    raise HTTPException(
+                        status_code=400, detail="File size exceeds 100MB"
+                    )
+
+                buffer.write(content)
+
+        except HTTPException:
+            raise
+
+        except Exception as e:
+            raise HTTPException(status_code=500, detail=f"Error saving file: {str(e)}")
+
+        return file_path, unique_filename
+
+
+
+    def upload_file(self, file: UploadFile):
+
+        self.validate_file(file)
+
+        file_path, stored_filename = self.save_file(file)
+
+        # Run cybersecurity static analysis automatically
+        analysis_result = analyze_file(file_path)
+
+        return {
+            "message": "File uploaded and analyzed successfully",
+            "original_filename": file.filename,
+            "stored_filename": stored_filename,
+            "file_size": os.path.getsize(file_path),
+            "upload_time": datetime.now(),
+            "analysis": analysis_result,
+        }

cybersecurity/file_analysis/__init__.py

@@ -0,0 +1,11 @@
+"""
+Static file analysis module.
+
+Supports:
+- EXE
+- DLL
+- ZIP
+- PDF
+- DOC
+- DOCX
+"""

cybersecurity/file_analysis/file_validator.py

@@ -0,0 +1,43 @@
+import os
+
+
+ALLOWED_EXTENSIONS = {
+    ".exe",
+    ".dll",
+    ".zip",
+    ".pdf",
+    ".doc",
+    ".docx",
+}
+
+MAX_FILE_SIZE = 100 * 1024 * 1024  # 100 MB
+
+
+def validate_file(file_path):
+    """
+    Validate that the file exists, has a supported extension,
+    and does not exceed the maximum allowed size.
+    """
+
+    if not os.path.isfile(file_path):
+        raise FileNotFoundError("File does not exist")
+
+    extension = os.path.splitext(file_path)[1].lower()
+
+    if extension not in ALLOWED_EXTENSIONS:
+        raise ValueError(
+            f"Unsupported file type: {extension}"
+        )
+
+    file_size = os.path.getsize(file_path)
+
+    if file_size > MAX_FILE_SIZE:
+        raise ValueError(
+            "File size exceeds 100 MB"
+        )
+
+    return {
+        "valid": True,
+        "extension": extension,
+        "size": file_size,
+    }

cybersecurity/file_analysis/hashing.py

@@ -0,0 +1,21 @@
+import hashlib
+
+
+def calculate_hashes(file_path):
+    """
+    Calculate MD5 and SHA-256 hashes.
+    """
+
+    md5 = hashlib.md5()
+    sha256 = hashlib.sha256()
+
+    with open(file_path, "rb") as file:
+
+        while chunk := file.read(4096):
+            md5.update(chunk)
+            sha256.update(chunk)
+
+    return {
+        "md5": md5.hexdigest(),
+        "sha256": sha256.hexdigest(),
+    }

cybersecurity/file_analysis/imports.py

@@ -0,0 +1,52 @@
+import pefile
+
+
+def extract_imports(file_path):
+    """
+    Extract imported DLLs and Windows API/function names
+    from a PE file.
+
+    This function only extracts names.
+    It does NOT classify APIs as suspicious or malicious.
+    """
+
+    pe = pefile.PE(file_path)
+
+    dlls = []
+    apis = []
+
+    if not hasattr(pe, "DIRECTORY_ENTRY_IMPORT"):
+        pe.close()
+
+        return {
+            "dlls": [],
+            "apis": []
+        }
+
+    for entry in pe.DIRECTORY_ENTRY_IMPORT:
+
+        dll_name = entry.dll.decode(
+            "utf-8",
+            errors="ignore"
+        )
+
+        dlls.append(dll_name)
+
+        for function in entry.imports:
+
+            if function.name:
+                function_name = function.name.decode(
+                    "utf-8",
+                    errors="ignore"
+                )
+            else:
+                function_name = f"ordinal_{function.ordinal}"
+
+            apis.append(function_name)
+
+    pe.close()
+
+    return {
+        "dlls": dlls,
+        "apis": apis
+    }