sameersyed/Defence_FrameWork
0
1from flask import Flask, request, jsonify, render_template, session, redirect, url_for, make_response2from flask_cors import CORS3import pandas as pd4import numpy as np5from backend.analysis_engine import SecurityAnalysisEngine6from backend.database import Database7from backend.spam_detector_ml import ImageSpamDetector8from backend.image_security_analyzer import ImageSecurityAnalyzer9from backend.dataset_analyzer import DatasetAnalyzer10from backend.ml_risk_analyzer import MLDatasetRiskAnalyzer11from backend.adversarial_generator import AdversarialGenerator12from backend.defense_simulator import DefenseSimulator13from backend.robustness_scorer import RobustnessScorer14from backend.physical_world_analyzer import PhysicalWorldAnalyzer15from functools import wraps16 17app = Flask(__name__)18app.secret_key = 'change-this-secret-key-in-production-12345'19app.config['SESSION_TYPE'] = 'filesystem'20app.config['PERMANENT_SESSION_LIFETIME'] = 3600 # 1 hour21CORS(app)22 23# Initialize components24engine = SecurityAnalysisEngine('backend/knowledge_base.json')25db = Database()26spam_detector = ImageSpamDetector()27image_analyzer = ImageSecurityAnalyzer()28dataset_analyzer = DatasetAnalyzer()29ml_risk_analyzer = MLDatasetRiskAnalyzer()30adversarial_gen = AdversarialGenerator()31defense_sim = DefenseSimulator()32robustness_scorer = RobustnessScorer()33physical_analyzer = PhysicalWorldAnalyzer()34 35# Login required decorator36def login_required(f):37 @wraps(f)38 def decorated_function(*args, **kwargs):39 if 'user_id' not in session:40 return redirect(url_for('login'))41 return f(*args, **kwargs)42 return decorated_function43 44@app.route('/signup', methods=['GET', 'POST'])45def signup():46 """User signup"""47 if request.method == 'POST':48 username = request.form.get('username')49 email = request.form.get('email')50 password = request.form.get('password')51 52 if db.create_user(username, email, password):53 return redirect(url_for('login'))54 else:55 return render_template('signup.html', error='Username or email already exists')56 57 return render_template('signup.html')58 59@app.route('/login', methods=['GET', 'POST'])60def login():61 """User login"""62 if request.method == 'POST':63 username = request.form.get('username')64 password = request.form.get('password')65 66 user_id = db.verify_user(username, password)67 if user_id:68 session['user_id'] = user_id69 user_data = db.get_user_by_id(user_id)70 session['username'] = user_data[0]71 return redirect(url_for('ui'))72 else:73 return render_template('login.html', error='Invalid credentials')74 75 return render_template('login.html')76 77@app.route('/logout')78def logout():79 """Logout user"""80 session.clear()81 return redirect(url_for('login'))82 83@app.route('/')84def landing():85 """Serve landing page"""86 return render_template('landing.html')87 88@app.route('/ui')89@login_required90def ui():91 """Serve the chat interface at /ui endpoint"""92 return render_template('index.html', username=session.get('username'))93 94@app.route('/dashboard')95@login_required96def index():97 """Redirect to /ui"""98 return redirect(url_for('ui'))99 100@app.route('/api/chat', methods=['POST'])101@login_required102def chat():103 """Process chat messages, images, dataset info, and attack classification"""104 try:105 data = request.get_json()106 user_message = data.get('message', '').strip()107 image_data = data.get('image')108 model_type = data.get('model_type', 'general')109 110 # Store model type in session111 session['model_type'] = model_type112 113 # Handle image spam detection114 if image_data:115 analysis = image_analyzer.analyze_image(image_data)116 response_text = image_analyzer.generate_report(analysis)117 else:118 # Handle text-based security analysis119 if not user_message:120 return jsonify({'error': 'Empty message'}), 400121 122 if len(user_message) > 1000:123 return jsonify({'error': 'Message too long'}), 400124 125 # Enhanced analysis126 intent = engine.analyze_intent(user_message)127 threats = engine.classify_threats(intent['threats'])128 defenses = engine.recommend_defenses(intent['threats'])129 130 # Generate comprehensive response131 response_text = engine.generate_response(intent, threats, defenses)132 133 # Save threats to history134 for threat in threats:135 db.save_threat(136 session['user_id'],137 threat['name'],138 threat['risk_level'],139 model_type,140 intent.get('confidence', 75)141 )142 143 # Save to database144 db.save_chat(session['user_id'], user_message or 'Analysis request', response_text)145 146 return jsonify({147 'response': response_text148 })149 150 except Exception as e:151 return jsonify({'error': f'Internal error: {str(e)}'}), 500152 153@app.route('/api/knowledge', methods=['GET'])154@login_required155def get_knowledge():156 """Return available attack types and defenses"""157 return jsonify({158 'attacks': list(engine.kb['attack_types'].keys()),159 'defenses': list(engine.kb['defenses'].keys())160 })161 162@app.route('/api/summary', methods=['GET'])163@login_required164def get_summary():165 """Return attack vs defense summary table"""166 summary = []167 for attack_id, attack_data in engine.kb['attack_types'].items():168 defenses = []169 for defense_id in attack_data['defenses'][:3]: # Top 3 defenses170 if defense_id in engine.kb['defenses']:171 defense = engine.kb['defenses'][defense_id]172 defenses.append({173 'name': defense['name'],174 'complexity': defense['complexity']175 })176 177 summary.append({178 'attack': attack_data['name'],179 'risk': attack_data['risk_level'],180 'defenses': defenses181 })182 183 return jsonify({'summary': summary})184 185 186 187@app.route('/api/threat-history', methods=['GET'])188@login_required189def get_threat_history():190 """Get user's threat history"""191 threats = db.get_threat_history(session['user_id'], limit=10)192 history = []193 for threat in threats:194 history.append({195 'threat_type': threat[0],196 'risk_level': threat[1],197 'model_type': threat[2],198 'confidence': threat[3],199 'timestamp': threat[4]200 })201 return jsonify({'history': history})202 203@app.route('/api/analyze-dataset', methods=['POST'])204@login_required205def analyze_dataset():206 """Analyze uploaded dataset with risk detection and charts"""207 try:208 if 'file' not in request.files:209 return jsonify({'error': 'No file uploaded'}), 400210 211 file = request.files['file']212 if file.filename == '':213 return jsonify({'error': 'No file selected'}), 400214 215 if not file.filename.endswith('.csv'):216 return jsonify({'error': 'Only CSV files allowed'}), 400217 218 # Read CSV with error handling219 import io220 try:221 df = pd.read_csv(222 io.StringIO(file.stream.read().decode('utf-8')),223 on_bad_lines='skip',224 encoding='utf-8',225 sep=None,226 engine='python'227 )228 except Exception as e:229 file.stream.seek(0)230 try:231 df = pd.read_csv(232 io.StringIO(file.stream.read().decode('latin-1')),233 on_bad_lines='skip',234 sep=None,235 engine='python'236 )237 except Exception as e2:238 return jsonify({'error': f'CSV parsing failed: {str(e2)}'}), 400239 240 if len(df) == 0:241 return jsonify({'error': 'CSV file is empty'}), 400242 243 print(f"DataFrame loaded: {len(df)} rows, {len(df.columns)} columns")244 245 # Perform ML-based analysis246 ml_analysis = ml_risk_analyzer.analyze(df)247 248 print(f"ML Analysis complete: {len(ml_analysis['risks'])} risks found")249 250 # Generate basic charts (simplified)251 charts = {}252 try:253 if len(ml_analysis['risks']) > 0:254 charts = dataset_analyzer.generate_charts(df, {255 'risks': ml_analysis['risks'],256 'overall_risk': ml_analysis['overall_risk_score'],257 'basic_info': {258 'total_samples': len(df),259 'total_features': len(df.columns),260 'numeric_features': len(df.select_dtypes(include=[np.number]).columns),261 'categorical_features': len(df.select_dtypes(include=['object']).columns),262 'memory_usage': f"{df.memory_usage(deep=True).sum() / 1024:.2f} KB"263 },264 'class_distribution': {},265 'outliers': {},266 'duplicates': {},267 'missing_values': {},268 'statistics': {}269 })270 except Exception as chart_error:271 print(f"Chart generation error: {str(chart_error)}")272 charts = {}273 274 # Save to session275 session['last_analysis'] = {276 'analysis': ml_analysis,277 'charts': charts,278 'filename': file.filename279 }280 281 return jsonify({282 'analysis': ml_analysis,283 'charts': charts,284 'filename': file.filename285 })286 287 except Exception as e:288 import traceback289 print(f"Error in analyze_dataset: {str(e)}")290 print(traceback.format_exc())291 return jsonify({'error': f'Analysis error: {str(e)}'}), 500292 293@app.route('/api/download-report', methods=['GET'])294@login_required295def download_report():296 """Download HTML report of last analysis"""297 try:298 if 'last_analysis' not in session:299 return jsonify({'error': 'No analysis available. Please analyze a dataset first.'}), 400300 301 data = session['last_analysis']302 303 # Convert analysis dict to proper format if needed304 if isinstance(data.get('analysis'), str):305 import json306 data['analysis'] = json.loads(data['analysis'])307 if isinstance(data.get('charts'), str):308 import json309 data['charts'] = json.loads(data['charts'])310 311 html_report = dataset_analyzer.generate_report_html(data['analysis'], data['charts'])312 313 response = make_response(html_report)314 response.headers['Content-Type'] = 'text/html'315 response.headers['Content-Disposition'] = f'attachment; filename=dataset_report_{data.get("filename", "analysis")}.html'316 317 return response318 319 except Exception as e:320 return jsonify({'error': f'Report generation error: {str(e)}'}), 500321 322@app.route('/api/generate-adversarial', methods=['POST'])323@login_required324def generate_adversarial():325 """Generate FGSM adversarial example"""326 try:327 data = request.get_json()328 image_data = data.get('image')329 epsilon = data.get('epsilon', 0.01)330 331 if not image_data:332 return jsonify({'error': 'No image provided'}), 400333 334 result = adversarial_gen.generate_fgsm(image_data, epsilon)335 336 if result['success']:337 report = adversarial_gen.generate_report(result)338 db.save_chat(session['user_id'], 'Adversarial Attack Generation', report)339 return jsonify({'result': result, 'report': report})340 else:341 return jsonify({'error': result.get('error', 'Generation failed')}), 500342 343 except Exception as e:344 return jsonify({'error': f'Generation error: {str(e)}'}), 500345 346@app.route('/api/apply-defense', methods=['POST'])347@login_required348def apply_defense():349 """Apply defense mechanism to image"""350 try:351 data = request.get_json()352 image_data = data.get('image')353 defense_type = data.get('defense_type', 'jpeg_compression')354 355 if not image_data:356 return jsonify({'error': 'No image provided'}), 400357 358 result = defense_sim.apply_defense(image_data, defense_type)359 360 if result['success']:361 report = defense_sim.generate_report(result)362 db.save_chat(session['user_id'], f'Defense Simulation: {defense_type}', report)363 return jsonify({'result': result, 'report': report})364 else:365 return jsonify({'error': result.get('error', 'Defense failed')}), 500366 367 except Exception as e:368 return jsonify({'error': f'Defense error: {str(e)}'}), 500369 370@app.route('/api/robustness-score', methods=['POST'])371@login_required372def calculate_robustness():373 """Calculate system robustness score"""374 try:375 data = request.get_json()376 analysis_results = data.get('analysis_results', {})377 378 score_result = robustness_scorer.calculate_score(analysis_results)379 report = robustness_scorer.generate_report(score_result)380 381 db.save_chat(session['user_id'], 'Robustness Score Calculation', report)382 383 return jsonify({'score': score_result, 'report': report})384 385 except Exception as e:386 return jsonify({'error': f'Scoring error: {str(e)}'}), 500387 388@app.route('/api/physical-world-analysis', methods=['POST'])389@login_required390def physical_world_analysis():391 """Analyze physical-world attack scenario"""392 try:393 data = request.get_json()394 scenario_id = data.get('scenario_id')395 396 if not scenario_id:397 return jsonify({'error': 'No scenario specified'}), 400398 399 analysis = physical_analyzer.analyze_scenario(scenario_id)400 401 if 'error' not in analysis:402 report = physical_analyzer.generate_report(analysis)403 db.save_chat(session['user_id'], f'Physical-World Analysis: {scenario_id}', report)404 return jsonify({'analysis': analysis, 'report': report})405 else:406 return jsonify({'error': analysis['error']}), 400407 408 except Exception as e:409 return jsonify({'error': f'Analysis error: {str(e)}'}), 500410 411@app.route('/api/physical-scenarios', methods=['GET'])412@login_required413def get_physical_scenarios():414 """Get list of physical-world attack scenarios"""415 scenarios = physical_analyzer.get_all_scenarios()416 return jsonify({'scenarios': scenarios})417 418if __name__ == '__main__':419 print("\n" + "="*60)420 print("ShieldML - Deep Learning Security Analysis System")421 print("="*60)422 print("Server starting on port 7860...")423 print("="*60 + "\n")424 app.run(debug=False, host='0.0.0.0', port=7860)425