Merge pull request #413 from JustGuardian/master

kimocoder · web-flow · commit fd8ec3dc01b8 · 2025-10-22T16:13:52.000+02:00
add --update-db option to update local MAC address prefix database
diff --git a/tools/fetch_oui.py b/tools/fetch_oui.py
@@ -0,0 +1,96 @@
+#!/usr/bin/env python3
+import argparse
+import csv
+import requests
+import sys
+import os
+from datetime import datetime
+
+# IEEE registry URLs
+IEEE_REG_URLS = {
+    "OUI":   "https://standards-oui.ieee.org/oui/oui.csv",
+    "MAM":   "https://standards-oui.ieee.org/oui28/mam.csv",
+    "OUI36": "https://standards-oui.ieee.org/oui36/oui36.csv",
+    "IAB":   "https://standards-oui.ieee.org/iab/iab.csv",
+}
+
+DEFAULT_FILENAME = "ieee-oui.txt"
+
+
+def fetch_csv(url, verbose=False):
+    """Download CSV content from a URL."""
+    headers = {"User-Agent": "Mozilla/5.0 (compatible; FetchOUI/1.0; +https://github.com/kimocoder/wifite2)"}
+    if verbose:
+        print(f"→ Fetching {url}")
+    response = requests.get(url, headers=headers, timeout=30)
+    if not response.ok:
+        raise RuntimeError(f"Failed to fetch {url}: {response.status_code} {response.reason}")
+    if len(response.content) == 0:
+        raise RuntimeError(f"Empty response from {url}")
+    if verbose:
+        print(f"  Downloaded {len(response.content)} bytes")
+    return response.text
+
+
+def parse_and_write_csv(csv_content, outfile, key, verbose=False):
+    """Parse CSV content and write MAC/Vendor to file."""
+    reader = csv.DictReader(csv_content.splitlines())
+    outfile.write(f"\n#\n# Start of IEEE {key} registry data\n#\n")
+    count = 0
+
+    for row in reader:
+        # Columns differ slightly between registries, so we handle gracefully
+        mac = row.get("Assignment") or row.get("Registry") or ""
+        vendor = row.get("Organization Name") or row.get("Organization") or ""
+        vendor = vendor.strip()
+        if mac and vendor:
+            outfile.write(f"{mac}\t{vendor}\n")
+            count += 1
+
+    outfile.write(f"#\n# End of IEEE {key} registry data. {count} entries.\n#\n")
+    if verbose:
+        print(f"  Wrote {count} entries for {key}")
+    return count
+
+
+def main():
+    parser = argparse.ArgumentParser(
+        description="Fetch the IEEE OUI (manufacturer) registries and write MAC/vendor mappings to a text file."
+    )
+    parser.add_argument("-f", metavar="FILE", default=DEFAULT_FILENAME,
+                        help=f"Output filename (default: {DEFAULT_FILENAME})")
+    parser.add_argument("-v", action="store_true", help="Verbose output")
+    args = parser.parse_args()
+
+    filename = args.f
+    verbose = args.v
+
+    # Delete old file if exists
+    if os.path.exists(filename):
+        if verbose:
+            print(f"Deleting existing {filename}")
+        os.remove(filename)
+
+    # Open new output file
+    if verbose:
+        print(f"Opening {filename} for output")
+
+    with open(filename, "w", encoding="utf-8") as outfile:
+        date_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+        outfile.write(f"# IEEE OUI Vendor List\n# Generated {date_str}\n")
+
+        total_entries = 0
+        for key, url in sorted(IEEE_REG_URLS.items()):
+            if verbose:
+                print(f"\nProcessing IEEE {key} registry data from {url}")
+            try:
+                content = fetch_csv(url, verbose)
+                total_entries += parse_and_write_csv(content, outfile, key, verbose)
+            except Exception as e:
+                print(f"Error processing {key}: {e}", file=sys.stderr)
+
+    print(f"\nTotal of {total_entries} MAC/Vendor mappings written to {filename}")
+
+
+if __name__ == "__main__":
+    main()
diff --git a/wifite/args.py b/wifite/args.py
@@ -518,6 +518,11 @@ def _add_command_args(commands):
                               action='store_true',
                               dest='crack_handshake',
                               help=Color.s('Show commands to crack a captured handshake'))
+        
+        commands.add_argument('--update-db',
+                              action='store_true',
+                              dest='update_db',
+                              help=Color.s('Update the local MAC address prefix database from IEEE registries'))
 
 
 if __name__ == '__main__':
diff --git a/wifite/config.py b/wifite/config.py
@@ -46,6 +46,8 @@ class Configuration(object):
     show_bssids = None
     show_cracked = None
     show_ignored = None
+    update_db = None
+    db_filename = None
     show_manufacturers = None
     skip_crack = None
     target_bssid = None
@@ -215,6 +217,8 @@ def initialize(cls, load_interface=True):
         cls.show_ignored = False
         cls.check_handshake = None
         cls.crack_handshake = False
+        cls.update_db = False
+        cls.db_filename = 'ieee-oui.txt'
 
         # A list to cache all checked commands (e.g. `which hashcat` will execute only once)
         cls.existing_commands = {}
@@ -267,7 +271,8 @@ def load_from_arguments(cls):
             cls.check_handshake = args.check_handshake
         if args.crack_handshake:
             cls.crack_handshake = True
-
+        if args.update_db:
+            cls.update_db = True
     @classmethod
     def validate(cls):
         if cls.use_pmkid_only and cls.wps_only:
diff --git a/wifite/util/dbupdater.py b/wifite/util/dbupdater.py
@@ -0,0 +1,113 @@
+#!/usr/bin/env python3
+import os
+import sys
+import csv
+import requests
+from datetime import datetime
+from typing import List
+
+# keep the same import style as your example (adjust to your project layout)
+from ..config import Configuration
+from ..util.color import Color
+from ..util.process import Process
+
+class DBUpdater:
+    """Updates a local database of MAC address prefixes to vendor names from IEEE registries.
+    """
+
+    # Registry URLs (same idea as your original script)
+    SOURCES = {
+        "OUI":   "https://standards-oui.ieee.org/oui/oui.csv",
+        "MAM":   "https://standards-oui.ieee.org/oui28/mam.csv",
+        "OUI36": "https://standards-oui.ieee.org/oui36/oui36.csv",
+        "IAB":   "https://standards-oui.ieee.org/iab/iab.csv",
+    }
+
+    DEFAULT_FILENAME = "ieee-oui.txt"
+
+    @classmethod
+    def run(cls):
+        
+        Configuration.initialize(False)
+
+        filename = Configuration.db_filename
+        verbose = bool(Configuration.verbose)
+
+        if os.path.exists(filename):
+            up_to_date, last_updated = cls.is_up_to_date(filename)
+            
+            if up_to_date:
+                Color.pl('{+} {G}Database is up to date ({C}%s{G}). Last update date: {C}%s{W}' % (filename, last_updated))
+                return
+            if verbose:
+                Color.pl('{!} {O}Deleting existing {R}%s{W}' % filename)
+            os.remove(filename)
+
+        try:
+            total_written = cls.update_all(filename, verbose=verbose)
+        except KeyboardInterrupt:
+            Color.pl('\n{!} {O}Interrupted by user{W}')
+            return
+
+        Color.pl('\n\n{+} {G}Done{W} - Total entries written: {C}%d{W}' % total_written)
+
+    @ classmethod
+    def update_all(cls, filename: str, verbose: bool = False) -> int:
+        """Loop selected sources, fetch, parse and append to filename. Returns count written."""
+        written_total = 0
+        with open(filename, "w", encoding="utf-8") as outfile:
+            date_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+            outfile.write(f"# Registry Vendor List\n# Generated {date_str}\n")
+            for key in cls.SOURCES.keys():
+                url = cls.SOURCES.get(key)
+                Color.pl('\n{+} Processing {C}%s{W} from {O}%s{W}' % (key, url))
+                try:
+                    csv_content = cls.fetch_csv(url, verbose=verbose)
+                    written = cls.parse_and_write_csv(csv_content, outfile, key, verbose=verbose)
+                    written_total += written
+                except Exception as e:
+                    print(f"Error processing {key}: {e}", file=sys.stderr)
+        return written_total
+    
+    
+    @classmethod
+    def fetch_csv(cls, url: str, verbose: bool = False) -> str:
+        """Download CSV content (boilerplate; uses requests)."""
+        headers = {"User-Agent": "Mozilla/5.0 (compatible; FetchOUI/1.0; +https://github.com/kimocoder/wifite2)"}
+        if verbose:
+            Color.pl('  →  Fetching %s' % url)
+        response = requests.get(url, headers=headers, timeout=30)
+        if not response.ok:
+            raise RuntimeError(f"Failed to fetch {url}: {response.status_code} {response.reason}")
+        if len(response.content) == 0:
+            raise RuntimeError(f"Empty response from {url}")
+        return response.text
+
+    @classmethod
+    def parse_and_write_csv(cls, csv_content: str, outfile, key: str, verbose: bool = False) -> int:
+        """Parse CSV content and write MAC\tVendor lines to outfile (boilerplate)."""
+        reader = csv.DictReader(csv_content.splitlines())
+        outfile.write(f"\n#\n# Start of {key} registry data\n#\n")
+        count = 0
+        for row in reader:
+            mac = row.get("Assignment") or row.get("Registry") or ""
+            vendor = row.get("Organization Name") or row.get("Organization") or ""
+            vendor = (vendor or "").strip()
+            if mac and vendor:
+                outfile.write(f"{mac}\t{vendor}\n")
+                count += 1
+        outfile.write(f"#\n# End of {key} registry data. {count} entries.\n#\n")
+
+        Color.p('     Wrote {C}%d{W} entries from source: {C}%s{W}' % (count, key))
+        return count
+
+    def is_up_to_date(filename: str) -> bool:
+        
+        mtime = os.path.getmtime(filename)
+        last_update = datetime.fromtimestamp(mtime).strftime("%Y-%m-%d %H:%M:%S")
+        age_seconds = datetime.now().timestamp() - mtime
+        return age_seconds < (7 * 24 * 3600), last_update #if file is older than 7 days it is not up to date
+
+
+if __name__ == '__main__':
+    DBUpdater.run()
diff --git a/wifite/wifite.py b/wifite/wifite.py
@@ -43,7 +43,8 @@ def start(self):
         from .model.result import CrackResult
         from .model.handshake import Handshake
         from .util.crack import CrackHelper
-
+        from .util.dbupdater import DBUpdater
+        
         if Configuration.show_cracked:
             CrackResult.display('cracked')
 
@@ -55,6 +56,10 @@ def start(self):
 
         elif Configuration.crack_handshake:
             CrackHelper.run()
+            
+        elif Configuration.update_db:
+            
+            DBUpdater.run()
 
         else:
             Configuration.get_monitor_mode_interface()
