Deeps-2005/java-ssl-scanner
0
1import streamlit as st2import requests3import json4import zipfile5import io6import hashlib # For hashing ZIP file content7 8# --- Page Configuration ---9st.set_page_config(10 page_title="SSL/HTTPS Vulnerability Scanner",11 layout="centered", # "wide" can also be used for more space12 initial_sidebar_state="collapsed",13 menu_items={14 'Get Help': 'https://www.example.com/help',15 'Report a bug': "https://www.example.com/bug",16 'About': "# This is an SSL/HTTPS Vulnerability Scanner powered by AI."17 }18)19 20# --- Session State Initialization ---21def init_session_state():22 """Initializes default values for Streamlit's session state."""23 defaults = {24 # Input-specific states for single file/pasted code25 "uploaded_file_bytes": None,26 "uploaded_filename": None,27 "pasted_code_bytes": None,28 "pasted_code_filename": "pasted_code.java", # Consistent name for pasted code29 30 # Input-specific states for ZIP files31 "zip_file_raw_bytes": None,32 "zip_file_content_hash": None, # MD5 hash of the ZIP content33 "zip_file_name": None,34 35 # Track which tab's content is currently being displayed/processed36 "current_display_source_tab": None, # "upload_file", "paste_code", "zip_file"37 38 # Analysis/Patching results for single file/pasted code39 "analysis_report_items": None,40 "complete_sanitized_code_overall": None,41 "trigger_patch_for_active_file": False,42 "patched_code_result": None,43 "patch_logs_result": None, # This will now be a list of dicts44 45 # Keep track of the uploader's state to detect actual changes for uploaded file46 "last_uploaded_filename": None,47 "last_uploaded_file_size": None,48 49 # Analysis/Patching results for ZIP files (cached dictionaries)50 # {filename_in_zip: {"analysis_items": [...], "complete_sanitized_code": "..."}}51 "zip_analysis_cache": {}, 52 # {filename_in_zip: {"patched_code": "...", "patch_logs": "..."}}53 "zip_patch_cache": {},54 # Set of filenames in ZIP for which patching has been triggered by a button click55 "zip_patch_triggered_files": set() 56 }57 for key, value in defaults.items():58 if key not in st.session_state:59 st.session_state[key] = value60 61init_session_state()62 63 64# --- Application Header ---65st.title("🔐 SSL/HTTPS Vulnerability Scanner & Auto-Patcher")66st.markdown(67 """68 Analyze your Java code for insecure usage of **X509TrustManager**,69 **HostnameVerifier**, weak **cipher suites**, and more. Get suggestions and **auto-patch** options.70 Choose your input method below:71 """72)73 74# --- Helper Function to Display Analysis Report ---75def display_analysis_items(analysis_items, filename_for_key=""):76 """Displays analysis results in an expandable format."""77 # Check for the specific "No vulnerabilities found" message first78 if isinstance(analysis_items, list) and len(analysis_items) == 1 and \79 analysis_items[0].get("issue") == "No vulnerabilities found by analyzer." and \80 analysis_items[0].get("severity") == "INFO":81 st.success("✅ No SSL/HTTPS vulnerabilities found in this code. Good job!")82 st.info("No analysis results to display.")83 return84 85 if isinstance(analysis_items, list) and len(analysis_items) == 1 and analysis_items[0].get("severity") == "ERROR":86 item = analysis_items[0]87 st.error(f"**❗ Analysis Error**: {item.get('issue', 'Unknown error')}")88 if item.get('suggestion'):89 st.warning(f"**💡 Suggestion**: {item.get('suggestion')}")90 return91 92 # Filter out ERRORs and the specific "No vulnerabilities found" INFO message93 valid_issues = [item for item in analysis_items 94 if item.get("severity") != "ERROR" and item.get("issue") and 95 not (item.get("issue") == "No vulnerabilities found by analyzer." and item.get("severity") == "INFO")]96 97 if valid_issues:98 st.info(f"Found {len(valid_issues)} potential vulnerabilities. Expand each section for details:")99 for i, item in enumerate(valid_issues):100 severity = item.get('severity', 'UNKNOWN')101 # Ensure issue_summary is robust102 issue_full = item.get('issue', 'Vulnerability Details')103 issue_parts = issue_full.split(':', 1)104 issue_summary_text = issue_parts[-1].strip().split('-')[0].strip() if issue_parts else issue_full105 106 line_info = f" (Line {item.get('line', 'N/A')})" if item.get('line') else ""107 108 color_emoji = "🔴" if severity == "CRITICAL" else \109 "🔥" if severity == "HIGH" else \110 "🟡" if severity == "MEDIUM" else \111 "⚪" if severity == "UNKNOWN" else \112 "❗"113 114 expander_title = f"{color_emoji} **{severity}**: {issue_summary_text}{line_info}"115 # Ensure unique key by incorporating filename and index116 expander_key = f"expander_{filename_for_key.replace('.', '_').replace('/', '_')}_{i}"117 118 with st.expander(expander_title):119 st.markdown(f"**❗ Full Issue**: {item.get('issue', '-')}")120 st.markdown(f"**💡 Suggestion**: {item.get('suggestion', 'No suggestion available')}")121 if "sanitized_code" in item and item["sanitized_code"] is not None:122 st.markdown("**🔧 Code Snippet (Suggestion):**")123 st.code(item["sanitized_code"].strip(), language="java")124 125 126# --- Helper Function to call Backend API ---127def call_backend_api(endpoint: str, filename: str, code_bytes: bytes, timeout: int = 60):128 """Calls a backend API endpoint (analyze or patch) and handles common errors."""129 api_url = f"http://127.0.0.1:8000/{endpoint}" # Consider making this configurable for deployment130 files_data = {'file': (filename, code_bytes, 'application/java')}131 try:132 response = requests.post(api_url, files=files_data, timeout=timeout)133 if response.ok:134 return {"ok": True, "data": response.json()}135 else:136 st.error(f"❌ API call to '{endpoint}' failed for {filename}. Status: {response.status_code}")137 return {"ok": False, "error": f"API Error: {response.status_code} - {response.text}"}138 except requests.exceptions.Timeout:139 st.error(f"⏰ API call to '{endpoint}' timed out for {filename}.")140 return {"ok": False, "error": "Timeout"}141 except requests.exceptions.ConnectionError:142 st.error(f"🔌 Could not connect to the backend for {filename}. Please ensure the backend server is running.")143 return {"ok": False, "error": "Connection Error"}144 except Exception as e:145 st.error(f"An unexpected error occurred during API call to '{endpoint}' for {filename}: {e}")146 return {"ok": False, "error": str(e)}147 148 149# --- Helper Function to clear processing results for single file/pasted code ---150def clear_processing_results_single_file():151 """Clears analysis/patching results related to single file/pasted code."""152 st.session_state.analysis_report_items = None153 st.session_state.complete_sanitized_code_overall = None154 st.session_state.trigger_patch_for_active_file = False155 st.session_state.patched_code_result = None156 st.session_state.patch_logs_result = None157 st.session_state.last_analyzed_filename = None158 st.session_state.last_analyzed_bytes = None159 160 161# --- Main processing function for single file/pasted code ---162def process_and_display_single_file(filename: str, code_bytes: bytes, source_tab_name: str):163 """164 Handles analysis and patching for a single Java file or pasted code.165 Caches results in session state to prevent redundant API calls.166 """167 # Only process if this is the content currently designated for display168 if st.session_state.current_display_source_tab != source_tab_name:169 return170 171 st.markdown(f"---")172 st.subheader(f"Processing: {filename}")173 174 # --- Analysis Section ---175 # Perform analysis only if results are not already in session state for the current active file176 # or if the file/source has changed.177 # We need to store the filename and bytes that were *last analyzed* to avoid re-analyzing.178 if st.session_state.analysis_report_items is None or \179 st.session_state.get("last_analyzed_filename") != filename or \180 st.session_state.get("last_analyzed_bytes") != code_bytes: # Compare bytes for pasted code181 182 clear_processing_results_single_file() # Clear previous results before new analysis183 st.session_state.last_analyzed_filename = filename184 st.session_state.last_analyzed_bytes = code_bytes # Store bytes for comparison185 186 with st.spinner(f"Analyzing {filename} for vulnerabilities..."):187 analysis_result = call_backend_api("analyze", filename, code_bytes, timeout=30)188 if analysis_result["ok"]:189 results_payload = analysis_result["data"].get("report", [])190 temp_analysis_items = []191 for item in results_payload:192 if "complete_sanitized_code" in item:193 st.session_state.complete_sanitized_code_overall = item["complete_sanitized_code"]194 else:195 temp_analysis_items.append(item)196 st.session_state.analysis_report_items = temp_analysis_items if temp_analysis_items else [{"issue": "No vulnerabilities found by analyzer.", "severity": "INFO"}]197 else:198 st.session_state.analysis_report_items = [{"issue": f"Analysis Failed: {analysis_result['error']}", "severity": "ERROR"}]199 200 # Display analysis results from session state201 if st.session_state.analysis_report_items is not None:202 display_analysis_items(st.session_state.analysis_report_items, filename)203 if st.session_state.complete_sanitized_code_overall:204 st.subheader(f"✅ Complete Auto-Patched Version for {filename} (from Analyzer)")205 st.code(st.session_state.complete_sanitized_code_overall.strip(), language="java")206 207 # --- Auto-Patch Section ---208 # Only show patch button if analysis found actual vulnerabilities (not just the INFO message or an ERROR)209 can_attempt_patch = st.session_state.analysis_report_items is not None and \210 any(item.get("severity") not in ["ERROR", "INFO"] for item in st.session_state.analysis_report_items)211 212 if can_attempt_patch:213 st.markdown(f"#### ⚙️ Auto-Patch Code for: {filename}")214 patch_button_key = f"patch_button_active_{filename.replace('.', '_').replace('/', '_').replace(' ', '_')}_{source_tab_name}"215 216 if st.button(f"Generate Patched Code for {filename}", key=patch_button_key):217 st.session_state.trigger_patch_for_active_file = True218 st.session_state.patched_code_result = None 219 st.session_state.patch_logs_result = None220 st.rerun() # Rerun to trigger the patching logic in the next script execution221 222 # Execute patching logic if triggered223 if st.session_state.trigger_patch_for_active_file:224 with st.spinner(f"Attempting to auto-patch {filename}..."):225 patch_result = call_backend_api("patch", filename, code_bytes, timeout=60)226 if patch_result["ok"]:227 patch_data = patch_result["data"]228 st.session_state.patched_code_result = patch_data.get("patched_code", "")229 st.session_state.patch_logs_result = patch_data.get("patch_logs", []) # Ensure it's a list230 if not st.session_state.patched_code_result and not st.session_state.patch_logs_result:231 st.session_state.patch_logs_result = [{"message": "Patching service returned an empty response. No changes made or no applicable patches found.", "line": "N/A"}]232 else:233 st.session_state.patched_code_result = "" # Ensure no old code is shown234 st.session_state.patch_logs_result = [{"message": f"Patching failed: {patch_result['error']}", "line": "N/A"}]235 236 # Display patched code and logs after patching is complete237 if st.session_state.patched_code_result:238 st.subheader(f"🧰 Patched Java Code Output for {filename}")239 st.code(st.session_state.patched_code_result.strip(), language="java")240 st.success("Auto-patching completed! Review the patched code above.")241 st.info("Remember: Automated patches may require manual review and testing.")242 243 # Display patch logs244 if st.session_state.patch_logs_result:245 st.subheader(f"📝 Patch Logs for {filename}")246 # Check if it's a list (expected) or a string (fallback from previous errors)247 if isinstance(st.session_state.patch_logs_result, list):248 if st.session_state.patch_logs_result:249 for log_entry in st.session_state.patch_logs_result:250 line = log_entry.get("line", "N/A")251 message = log_entry.get("message", "No message provided.")252 st.text_area(f"Line {line}", value=message, height=70, disabled=True, key=f"log_display_{filename}_{line}_{message[:20]}")253 else:254 st.info("No specific patch logs were generated.")255 else: # Fallback for unexpected string format256 st.text_area("Raw Patch Logs (Unexpected Format):", value=str(st.session_state.patch_logs_result), height=100, disabled=True)257 258 # Reset trigger after display259 st.session_state.trigger_patch_for_active_file = False260 st.session_state.patch_logs_result = None # Clear logs after display261 262# --- Streamlit Tabs for Input Method ---263tab1, tab2, tab3 = st.tabs(["Upload .java File", "Paste Java Code", "Upload ZIP File (Multiple Files)"])264 265# --- Tab 1: Upload .java File ---266with tab1:267 st.subheader("Upload a Single Java (.java) File")268 uploaded_file_single = st.file_uploader("Choose a .java file", type=["java"], key="single_file_uploader")269 270 # Clear results if a new file is uploaded271 if uploaded_file_single and (st.session_state.last_uploaded_filename != uploaded_file_single.name or \272 st.session_state.last_uploaded_file_size != uploaded_file_single.size):273 clear_processing_results_single_file()274 st.session_state.uploaded_file_bytes = uploaded_file_single.read()275 st.session_state.uploaded_filename = uploaded_file_single.name276 st.session_state.last_uploaded_filename = uploaded_file_single.name277 st.session_state.last_uploaded_file_size = uploaded_file_single.size278 st.session_state.current_display_source_tab = "upload_file"279 st.rerun() # Rerun to process the new file280 281 # If the file exists and is the current active display source, process it282 if st.session_state.current_display_source_tab == "upload_file" and \283 st.session_state.uploaded_file_bytes is not None and \284 st.session_state.uploaded_filename is not None:285 process_and_display_single_file(st.session_state.uploaded_filename, st.session_state.uploaded_file_bytes, "upload_file")286 elif uploaded_file_single is None and st.session_state.current_display_source_tab == "upload_file":287 clear_processing_results_single_file() # Clear if user removed the file288 289# --- Tab 2: Paste Java Code ---290with tab2:291 st.subheader("Paste Your Java Code Here")292 pasted_code_input = st.text_area("Paste code...", height=300, key="pasted_code_area")293 294 # Check if pasted code has changed and update session state295 if pasted_code_input:296 current_pasted_bytes = pasted_code_input.encode("utf-8")297 if st.session_state.pasted_code_bytes != current_pasted_bytes:298 clear_processing_results_single_file()299 st.session_state.pasted_code_bytes = current_pasted_bytes300 st.session_state.current_display_source_tab = "paste_code"301 st.rerun() # Rerun to process new pasted code302 elif not pasted_code_input and st.session_state.current_display_source_tab == "paste_code":303 clear_processing_results_single_file() # Clear if user cleared the text area304 305 # If pasted code exists and is the current active display source, process it306 if st.session_state.current_display_source_tab == "paste_code" and \307 st.session_state.pasted_code_bytes is not None:308 process_and_display_single_file(st.session_state.pasted_code_filename, st.session_state.pasted_code_bytes, "paste_code")309 310 311# --- Tab 3: Upload ZIP File ---312with tab3:313 st.subheader("Upload a ZIP File containing .java files")314 zip_file_upload = st.file_uploader("Choose a .zip file", type=["zip"], key="zip_file_uploader")315 316 if zip_file_upload:317 raw_bytes = zip_file_upload.read()318 current_hash = hashlib.md5(raw_bytes).hexdigest()319 320 # Check if a new ZIP file is uploaded or content changed321 if st.session_state.zip_file_content_hash != current_hash:322 st.session_state.zip_file_raw_bytes = raw_bytes323 st.session_state.zip_file_content_hash = current_hash324 st.session_state.zip_file_name = zip_file_upload.name325 st.session_state.zip_analysis_cache = {} # Clear cache for new zip326 st.session_state.zip_patch_cache = {}327 st.session_state.zip_patch_triggered_files = set()328 st.session_state.current_display_source_tab = "zip_file" # Set active tab329 st.rerun() # Rerun to process the new zip330 331 st.markdown(f"**Processing ZIP**: `{st.session_state.zip_file_name}`")332 st.info("Results for each Java file within the ZIP will be displayed below.")333 334 try:335 with zipfile.ZipFile(io.BytesIO(st.session_state.zip_file_raw_bytes), 'r') as zf:336 java_files_in_zip = [name for name in zf.namelist() if name.lower().endswith('.java') and not name.startswith('__MACOSX/')]337 338 if not java_files_in_zip:339 st.warning("No .java files found in the uploaded ZIP archive.")340 st.session_state.zip_analysis_cache = {} # Ensure empty if no java files341 st.session_state.zip_patch_cache = {}342 else:343 for member_name in sorted(java_files_in_zip): # Sort for consistent display order344 st.markdown(f"---")345 st.markdown(f"### File: `{member_name}`")346 347 # --- ZIP File Analysis ---348 # Only analyze if not already cached349 if member_name not in st.session_state.zip_analysis_cache:350 st.subheader("Scanning for Vulnerabilities...")351 with st.spinner(f"Analyzing `{member_name}`..."):352 member_content = zf.read(member_name)353 analysis_result = call_backend_api("analyze", member_name, member_content, timeout=30)354 if analysis_result["ok"]:355 st.session_state.zip_analysis_cache[member_name] = {356 "analysis_items": analysis_result["data"].get("report", []),357 # "complete_sanitized_code": analysis_result["data"].get("complete_sanitized_code", "") # Analyzer doesn't send this358 }359 else:360 st.session_state.zip_analysis_cache[member_name] = {361 "analysis_items": [{"issue": f"Analysis Failed: {analysis_result['error']}", "severity": "ERROR"}],362 }363 364 # Display analysis results for the current ZIP member365 if member_name in st.session_state.zip_analysis_cache:366 display_analysis_items(st.session_state.zip_analysis_cache[member_name]["analysis_items"], member_name)367 368 # --- ZIP File Patching ---369 # Only show patch button if analysis found actual vulnerabilities for this file370 member_analysis_items = st.session_state.zip_analysis_cache[member_name]["analysis_items"]371 can_patch_zip_member = any(item.get("severity") not in ["ERROR", "INFO"] for item in member_analysis_items)372 373 if can_patch_zip_member:374 st.markdown(f"#### ⚙️ Auto-Patch Code for: `{member_name}`")375 patch_button_key_zip = f"patch_button_zip_{member_name.replace('.', '_').replace('/', '_')}"376 377 if st.button(f"Generate Patched Code for {member_name}", key=patch_button_key_zip):378 st.session_state.zip_patch_triggered_files.add(member_name)379 st.session_state.zip_patch_cache[member_name] = {"patched_code": None, "patch_logs": None} # Reset before patch380 st.rerun() # Trigger rerun for patching381 382 # Execute patching logic if triggered for this file383 if member_name in st.session_state.zip_patch_triggered_files:384 if st.session_state.zip_patch_cache[member_name].get("patched_code") is None: # Only run if not already patched/cached385 with st.spinner(f"Attempting to auto-patch `{member_name}`..."):386 member_content = zf.read(member_name) # Re-read content387 patch_result = call_backend_api("patch", member_name, member_content, timeout=60)388 if patch_result["ok"]:389 patch_data = patch_result["data"]390 st.session_state.zip_patch_cache[member_name]["patched_code"] = patch_data.get("patched_code", "")391 st.session_state.zip_patch_cache[member_name]["patch_logs"] = patch_data.get("patch_logs", []) # Ensure it's a list392 if not st.session_state.zip_patch_cache[member_name]["patched_code"] and not st.session_state.zip_patch_cache[member_name]["patch_logs"]:393 st.session_state.zip_patch_cache[member_name]["patch_logs"] = [{"message": "Patching service returned an empty response. No changes made or no applicable patches found.", "line": "N/A"}]394 else:395 st.session_state.zip_patch_cache[member_name]["patched_code"] = ""396 st.session_state.zip_patch_cache[member_name]["patch_logs"] = [{"message": f"Patching failed: {patch_result['error']}", "line": "N/A"}]397 398 # Display patched code and logs for the current ZIP member399 patched_code_display = st.session_state.zip_patch_cache[member_name]["patched_code"]400 patch_logs_display = st.session_state.zip_patch_cache[member_name]["patch_logs"]401 402 if patched_code_display:403 st.subheader(f"🧰 Patched Code Output for {member_name}")404 st.code(patched_code_display.strip(), language="java")405 st.success(f"Auto-patching completed for {member_name}! Review the patched code.")406 else:407 st.warning(f"No patched code was returned for {member_name}.")408 409 # Display patch logs for ZIP file members410 if patch_logs_display:411 st.subheader(f"📝 Patch Logs for {member_name}")412 if isinstance(patch_logs_display, list):413 if patch_logs_display:414 for log_entry in patch_logs_display:415 line = log_entry.get("line", "N/A")416 message = log_entry.get("message", "No message provided.")417 st.text_area(f"Line {line}", value=message, height=70, disabled=True, key=f"log_zip_display_{member_name}_{line}_{message[:20]}")418 else:419 st.info(f"No specific patch logs were generated for {member_name}.")420 else: # Fallback for unexpected string format421 st.text_area("Raw Patch Logs (Unexpected Format):", value=str(patch_logs_display), height=100, disabled=True)422 else:423 st.warning(f"No analysis results available for {member_name} to patch, or no vulnerabilities found.")424 425 426 except zipfile.BadZipFile:427 st.error("The uploaded file is not a valid ZIP archive or is corrupted.")428 # Clear ZIP state on bad file429 st.session_state.zip_file_raw_bytes = None430 st.session_state.zip_file_content_hash = None431 st.session_state.zip_file_name = None432 st.session_state.zip_analysis_cache = {}433 st.session_state.zip_patch_cache = {}434 st.session_state.zip_patch_triggered_files = set()435 except Exception as e:436 st.error(f"An error occurred while processing the ZIP file: {e}")437 st.exception(e)438 # Clear ZIP state on error439 st.session_state.zip_file_raw_bytes = None440 st.session_state.zip_file_content_hash = None441 st.session_state.zip_file_name = None442 st.session_state.zip_analysis_cache = {}443 st.session_state.zip_patch_cache = {}444 st.session_state.zip_patch_triggered_files = set()