Update app.py

app.py

@@ -15,88 +15,93 @@ from collections import defaultdict
 urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
 warnings.filterwarnings('ignore')
 
+# GitHub raw URL for the ports file
+GITHUB_PORTS_RAW_URL = "https://raw.githubusercontent.com/Wuhpondiscord/ports/main/common_ports.txt"
+
 # --- Enhanced Port Parsing with Multi-Port Service Detection ---
 
-def load_custom_ports(folder_path="ports") -> Dict[int, str]:
+def load_custom_ports() -> Dict[int, str]:
     """
     Enhanced parser that:
     1. Handles continuous format: '135 - MS SQL1433 - MSSQL1521 - Oracle'
     2. Detects services on multiple ports
     3. Groups related services
+    Loads port definitions from a specified GitHub raw URL.
     """
     port_map = {80: "HTTP", 443: "HTTPS"}
     service_ports = defaultdict(list)  # Track which ports belong to same service
     
-    if not os.path.exists(folder_path):
-        os.makedirs(folder_path)
-        return port_map
+    try:
+        print(f"🌐 Fetching port definitions from: {GITHUB_PORTS_RAW_URL}")
+        response = requests.get(GITHUB_PORTS_RAW_URL, timeout=10)
+        response.raise_for_status()  # Raise an HTTPError for bad responses (4xx or 5xx)
+        content = response.text
+        print("✅ Successfully fetched port definitions.")
 
-    for filename in os.listdir(folder_path):
-        if filename.endswith(('.txt', '.csv', '.conf', '.list')):
-            file_path = os.path.join(folder_path, filename)
+        # Strategy 1: Continuous format parser
+        continuous_pattern = r'(\d+)\s*-\s*([A-Za-z][\w\s\-\.\(\)]*?)(?=\d+\s*-|$)'
+        matches = re.findall(continuous_pattern, content, re.MULTILINE)
+        
+        for port_str, desc in matches:
             try:
-                with open(file_path, 'r', encoding='utf-8', errors='ignore') as f:
-                    content = f.read()
-                
-                # Strategy 1: Continuous format parser
-                continuous_pattern = r'(\d+)\s*-\s*([A-Za-z][\w\s\-\.\(\)]*?)(?=\d+\s*-|$)'
-                matches = re.findall(continuous_pattern, content, re.MULTILINE)
-                
-                for port_str, desc in matches:
+                port_num = int(port_str)
+                if 1 <= port_num <= 65535:
+                    desc = desc.strip().strip('-').strip()
+                    desc = re.sub(r'\s+', ' ', desc)
+                    if desc:
+                        # Extract base service name
+                        base_service = desc.split('-')[0].strip()
+                        service_ports[base_service].append(port_num)
+                        port_map[port_num] = desc
+            except ValueError:
+                continue
+        
+        # Strategy 2: Line-by-line format
+        for line in content.split('\n'):
+            line = line.strip()
+            if not line or line.startswith('#'):
+                continue
+            
+            # Skip if already matched by continuous pattern
+            if re.match(r'^\d+\s*-\s*[A-Za-z]', line):
+                continue
+            
+            patterns = [
+                r'^(.+?)\s*[:=]\s*(\d+)', # e.g., MySQL: 3306
+                r'port\s*[:=]?\s*(\d+)\s+(.+)', # e.g., port 22 SSH
+                r'(\d+)\s*/\s*\w+\s+(.+)', # e.g., 22/tcp SSH
+                r'^(\d+)\s+(.+)$', # e.g., 22 SSH
+            ]
+            
+            for pattern in patterns:
+                match = re.match(pattern, line, re.IGNORECASE)
+                if match:
+                    groups = match.groups()
+                    
+                    if groups[0].isdigit():
+                        port_str, desc = groups[0], groups[1]
+                    else:
+                        desc, port_str = groups[0], groups[1]
+                    
                     try:
                         port_num = int(port_str)
                         if 1 <= port_num <= 65535:
-                            desc = desc.strip().strip('-').strip()
+                            desc = re.sub(r'[^\w\s\-\.]+', '', desc).strip()
                             desc = re.sub(r'\s+', ' ', desc)
                             if desc:
-                                # Extract base service name
                                 base_service = desc.split('-')[0].strip()
                                 service_ports[base_service].append(port_num)
                                 port_map[port_num] = desc
+                            break
                     except ValueError:
                         continue
-                
-                # Strategy 2: Line-by-line format
-                for line in content.split('\n'):
-                    line = line.strip()
-                    if not line or line.startswith('#'):
-                        continue
-                    
-                    if re.match(r'^\d+\s*-\s*[A-Za-z]', line):
-                        continue
-                    
-                    patterns = [
-                        r'^(.+?)\s*[:=]\s*(\d+)',
-                        r'port\s*[:=]?\s*(\d+)\s+(.+)',
-                        r'(\d+)\s*/\s*\w+\s+(.+)',
-                        r'^(\d+)\s+(.+)$',
-                    ]
-                    
-                    for pattern in patterns:
-                        match = re.match(pattern, line, re.IGNORECASE)
-                        if match:
-                            groups = match.groups()
-                            
-                            if groups[0].isdigit():
-                                port_str, desc = groups[0], groups[1]
-                            else:
-                                desc, port_str = groups[0], groups[1]
                             
-                            try:
-                                port_num = int(port_str)
-                                if 1 <= port_num <= 65535:
-                                    desc = re.sub(r'[^\w\s\-\.]+', '', desc).strip()
-                                    desc = re.sub(r'\s+', ' ', desc)
-                                    if desc:
-                                        base_service = desc.split('-')[0].strip()
-                                        service_ports[base_service].append(port_num)
-                                        port_map[port_num] = desc
-                                    break
-                            except ValueError:
-                                continue
-                                
-            except Exception as e:
-                print(f"⚠️ Error reading {filename}: {e}")
+    except requests.exceptions.RequestException as e:
+        print(f"⚠️ Error fetching port definitions from GitHub: {e}")
+        print("Using default port definitions.")
+    except Exception as e:
+        print(f"⚠️ Error parsing fetched port definitions: {e}")
+        print("Using default port definitions.")
     
     # Add multi-port info to descriptions
     for service, ports in service_ports.items():
@@ -114,147 +119,9 @@ def load_custom_ports(folder_path="ports") -> Dict[int, str]:
     print(f"✅ Loaded {len(port_map)} port definitions")
     return port_map
 
+# --- Rest of your code remains the same ---
 
-# --- IP Extraction ---
-
-def extract_ips(file_path: str) -> List[str]:
-    """Enhanced IP extractor with validation."""
-    try:
-        with open(file_path, 'r', encoding='utf-8', errors='ignore') as f:
-            content = f.read()
-        
-        found_ips = set(re.findall(r'\b(?:\d{1,3}\.){3}\d{1,3}\b', content))
-        
-        valid_ips = []
-        for ip in found_ips:
-            octets = [int(x) for x in ip.split('.')]
-            
-            if any(o > 255 for o in octets):
-                continue
-            
-            if (octets[0] in [0, 127, 255] or
-                octets[0] == 10 or
-                (octets[0] == 172 and 16 <= octets[1] <= 31) or
-                (octets[0] == 192 and octets[1] == 168) or
-                (octets[0] == 169 and octets[1] == 254)):
-                continue
-            
-            valid_ips.append(ip)
-        
-        return sorted(valid_ips)
-    except Exception as e:
-        print(f"Error extracting IPs: {e}")
-        return []
-
-
-# --- C2 Detection ---
-
-C2_SIGNATURES = {
-    "Cobalt Strike": ["404 Not Found", "application/ocsp-response", "BeEF", "cobaltstrike"],
-    "Metasploit": ["Metasploit", "Mettle", "meterpreter"],
-    "Covenant": ["covenant", "GruntHTTP", "Auth/Login"],
-    "Empire": ["Empire", "session_id", "admin/login.php"],
-    "Sliver": ["sliver", "implant"],
-    "Mythic": ["mythic", "agent_message"]
-}
-
-
-def probe_target(ip: str, port: int, port_desc: str) -> Optional[Dict]:
-    """Enhanced probe with better banner detection and multiple protocol attempts."""
-    sock = None
-    try:
-        # Port connectivity check
-        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-        sock.settimeout(1.2)
-        result = sock.connect_ex((ip, port))
-        
-        if result != 0:
-            return None
-        
-        # Try to get raw banner first (works for many services)
-        banner = "N/A"
-        try:
-            sock.send(b'\r\n')
-            raw_banner = sock.recv(1024).decode('utf-8', errors='ignore').strip()
-            if raw_banner and len(raw_banner) > 3:
-                banner = raw_banner[:200]  # First 200 chars
-        except:
-            pass
-        finally:
-            if sock:
-                sock.close()
-                sock = None
-        
-        # Try HTTP/HTTPS
-        for protocol in ["https", "http"]:
-            # Smart protocol selection
-            if protocol == "https" and port not in [443, 8443, 2083, 2087, 9443, 8000, 4443]:
-                if port < 8000:  # Skip HTTPS for low ports unless common HTTPS ports
-                    continue
-            
-            try:
-                resp = requests.get(
-                    f"{protocol}://{ip}:{port}",
-                    timeout=1.8,
-                    verify=False,
-                    allow_redirects=False,
-                    headers={'User-Agent': 'Mozilla/5.0'}
-                )
-                
-                # Get server banner
-                server = resp.headers.get('Server', '')
-                powered_by = resp.headers.get('X-Powered-By', '')
-                
-                if server or powered_by:
-                    banner = f"{server} {powered_by}".strip()
-                elif banner == "N/A":
-                    banner = f"HTTP {resp.status_code}"
-                
-                content = resp.text[:5000]
-                
-                category = f"Web ({port_desc})"
-                
-                # C2 Detection
-                for name, sigs in C2_SIGNATURES.items():
-                    if any(sig.lower() in content.lower() or 
-                          sig.lower() in str(resp.headers).lower() for sig in sigs):
-                        category = f"🚨 POTENTIAL {name}"
-                        break
-                
-                return {
-                    "IP": ip,
-                    "Port": port,
-                    "Service": port_desc,
-                    "Type": category,
-                    "Banner": banner
-                }
-            except requests.exceptions.SSLError:
-                # SSL error on HTTP, try HTTPS
-                if protocol == "http":
-                    continue
-                else:
-                    break
-            except:
-                continue
-        
-        # Port is open but not HTTP
-        return {
-            "IP": ip,
-            "Port": port,
-            "Service": port_desc,
-            "Type": "Open (Non-HTTP)",
-            "Banner": banner if banner != "N/A" else "Unknown"
-        }
-        
-    except Exception as e:
-        return None
-    finally:
-        if sock:
-            try:
-                sock.close()
-            except:
-                pass
-
+# ... (IP Extraction, C2 Detection, Probe Target, Start Analysis, UI) ...
 
 def start_analysis(file_obj, max_threads: int = 50, progress=gr.Progress()):
     """Main analysis with pure threading."""
@@ -263,7 +130,8 @@ def start_analysis(file_obj, max_threads: int = 50, progress=gr.Progress()):
     
     try:
         progress(0, desc="📋 Parsing port definitions...")
-        port_map = load_custom_ports("ports")
+        # Call load_custom_ports without a folder_path argument
+        port_map = load_custom_ports() 
         
         progress(0.1, desc="🔍 Extracting IP addresses...")
         ips = extract_ips(file_obj.name)
@@ -396,8 +264,8 @@ with demo:
             )
         
         with gr.Column(scale=1):
-            gr.Markdown("""
-            ### 🔧 Port Format Support
+            gr.Markdown(f"""
+            ### 🔧 Port Format Support (from {GITHUB_PORTS_RAW_URL.split('/')[-1]})
             
             **Continuous** (your format):
             ```
@@ -442,26 +310,26 @@ with demo:
 
 
 if __name__ == "__main__":
-    os.makedirs("ports", exist_ok=True)
-    
-    sample_file = "ports/common_ports.txt"
-    if not os.path.exists(sample_file):
-        with open(sample_file, 'w') as f:
-            f.write("""# Common Ports
-21 - FTP
-22 - SSH
-23 - Telnet
-25 - SMTP
-80 - HTTP
-443 - HTTPS
-3306 - MySQL
-3307 - MySQL Alternate
-3389 - RDP
-5432 - PostgreSQL
-5433 - PostgreSQL Alternate
-8080 - HTTP Alt
-8443 - HTTPS Alt
-""")
+    # Removed local file creation, as we're now fetching from GitHub
+    # os.makedirs("ports", exist_ok=True)
+    # sample_file = "ports/common_ports.txt"
+    # if not os.path.exists(sample_file):
+    #     with open(sample_file, 'w') as f:
+    #         f.write("""# Common Ports
+    # 21 - FTP
+    # 22 - SSH
+    # 23 - Telnet
+    # 25 - SMTP
+    # 80 - HTTP
+    # 443 - HTTPS
+    # 3306 - MySQL
+    # 3307 - MySQL Alternate
+    # 3389 - RDP
+    # 5432 - PostgreSQL
+    # 5433 - PostgreSQL Alternate
+    # 8080 - HTTP Alt
+    # 8443 - HTTPS Alt
+    # """)
     
     demo.queue(max_size=10)
     demo.launch(