simon/RadMac
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

more changes

Browse Source
This commit is contained in:
Simon Cloutier
2025-04-02 14:52:23 -04:00
parent 82e534f4d3
commit bfd6d8af57
15 changed files with 601 additions and 293 deletions
Download Patch File Download Diff File Expand all files Collapse all files

174
app/db_interface.py
View File

@@ -18,12 +18,14 @@ def get_all_users():
conn = get_connection()
cursor = conn.cursor(dictionary=True)
cursor.execute("""
SELECT u.*, g.vlan_id AS group_vlan_id, g.description AS group_description,
mv.vendor_name
SELECT
u.*,
g.vlan_id AS group_vlan_id,
g.description AS group_description,
COALESCE(m.vendor_name, '...') AS vendor
FROM users u
LEFT JOIN groups g ON u.vlan_id = g.vlan_id
LEFT JOIN mac_vendors mv
ON SUBSTRING(REPLACE(REPLACE(u.mac_address, ':', ''), '-', ''), 1, 6) = mv.mac_prefix
LEFT JOIN mac_vendors m ON LOWER(REPLACE(REPLACE(u.mac_address, ':', ''), '-', '')) LIKE CONCAT(m.mac_prefix, '%')
""")
users = cursor.fetchall()
cursor.close()
@@ -32,6 +34,7 @@ def get_all_users():
def get_all_groups():
conn = get_connection()
cursor = conn.cursor(dictionary=True)
@@ -71,7 +74,7 @@ def add_group(vlan_id, description):
def update_group_description(vlan_id, description):
conn = get_connection()
cursor = conn.cursor()
cursor.execute("UPDATE groups SET description = %s WHERE id = %s", (description, vlan_id))
cursor.execute("UPDATE groups SET description = %s WHERE vlan_id = %s", (description, vlan_id))
conn.commit()
cursor.close()
conn.close()
@@ -80,7 +83,7 @@ def update_group_description(vlan_id, description):
def delete_group(vlan_id):
conn = get_connection()
cursor = conn.cursor()
cursor.execute("DELETE FROM groups WHERE id = %s", (vlan_id,))
cursor.execute("DELETE FROM groups WHERE vlan_id = %s", (vlan_id,))
conn.commit()
cursor.close()
conn.close()
@@ -89,7 +92,7 @@ def delete_group(vlan_id):
def duplicate_group(vlan_id):
conn = get_connection()
cursor = conn.cursor(dictionary=True)
cursor.execute("SELECT vlan_id, description FROM groups WHERE id = %s", (vlan_id,))
cursor.execute("SELECT vlan_id, description FROM groups WHERE vlan_id = %s", (vlan_id,))
group = cursor.fetchone()
if group:
@@ -154,18 +157,94 @@ def get_latest_auth_logs(result, limit=10):
return logs
def get_vendor_info(mac):
def get_vendor_info(mac, insert_if_found=True):
conn = get_connection()
cursor = conn.cursor(dictionary=True)
prefix = mac.lower().replace(":", "").replace("-", "")[:6]
cursor.execute("SELECT vendor FROM mac_vendors WHERE prefix = %s", (prefix,))
print(f">>> Looking up MAC: {mac} → Prefix: {prefix}")
print("→ Searching in local database...")
cursor.execute("SELECT vendor_name, status FROM mac_vendors WHERE mac_prefix = %s", (prefix,))
row = cursor.fetchone()
cursor.close()
conn.close()
return row['vendor'] if row else "Unknown Vendor"
if row:
print(f"✓ Found locally: {row['vendor_name']} (Status: {row['status']})")
cursor.close()
conn.close()
return {
"mac": mac,
"vendor": row['vendor_name'],
"source": "local",
"status": row['status']
}
print("✗ Not found locally, querying API...")
url_template = current_app.config.get("OUI_API_URL", "https://api.maclookup.app/v2/macs/{}")
api_key = current_app.config.get("OUI_API_KEY", "")
headers = {"Authorization": f"Bearer {api_key}"} if api_key else {}
try:
url = url_template.format(prefix)
print(f"→ Querying API: {url}")
response = requests.get(url, headers=headers)
if response.status_code == 200:
data = response.json()
vendor = data.get("company", "").strip()
if vendor:
print(f"✓ Found from API: {vendor}")
if insert_if_found:
cursor.execute("""
INSERT INTO mac_vendors (mac_prefix, vendor_name, status, last_checked, last_updated)
VALUES (%s, %s, 'found', NOW(), NOW())
ON DUPLICATE KEY UPDATE
vendor_name = VALUES(vendor_name),
status = 'found',
last_checked = NOW(),
last_updated = NOW()
""", (prefix, vendor))
conn.commit()
print("→ Inserted into database (found).")
return {
"mac": mac,
"vendor": vendor,
"source": "api",
"status": "found"
}
elif response.status_code == 404:
print("✗ API returned 404 - vendor not found.")
if insert_if_found:
cursor.execute("""
INSERT INTO mac_vendors (mac_prefix, vendor_name, status, last_checked, last_updated)
VALUES (%s, %s, 'not_found', NOW(), NOW())
ON DUPLICATE KEY UPDATE
vendor_name = VALUES(vendor_name),
status = 'not_found',
last_checked = NOW(),
last_updated = NOW()
""", (prefix, "not found"))
conn.commit()
print("→ Inserted into database (not_found).")
return {
"mac": mac,
"vendor": "",
"source": "api",
"status": "not_found"
}
else:
print(f"✗ API error: {response.status_code}")
return {"mac": mac, "vendor": "", "error": f"API error: {response.status_code}"}
except Exception as e:
print(f"✗ Exception while querying API: {e}")
return {"mac": mac, "vendor": "", "error": str(e)}
finally:
cursor.close()
conn.close()
def get_summary_counts():
conn = get_connection()
@@ -239,16 +318,22 @@ def refresh_vendors():
try:
url = url_template.format(prefix)
response = requests.get(url, headers=headers)
if response.status_code == 200:
data = response.json()
vendor_name = data.get("company", "not found")
status = "found"
vendor_name = data.get("company", "").strip()
if vendor_name:
status = "found"
else:
# Empty "company" field — skip insert
print(f"⚠️ Empty vendor for {prefix}, skipping.")
continue
elif response.status_code == 404:
vendor_name = "not found"
status = "not_found"
else:
print(f"Error {response.status_code} for {prefix}")
continue # skip insert on unexpected status
print(f"❌ API error {response.status_code} for {prefix}, skipping.")
continue
cursor.execute("""
INSERT INTO mac_vendors (mac_prefix, vendor_name, status, last_checked, last_updated)
@@ -263,10 +348,65 @@ def refresh_vendors():
inserted += 1
except Exception as e:
print(f"Error fetching vendor for {prefix}: {e}")
print(f"🚨 Exception fetching vendor for {prefix}: {e}")
continue
time.sleep(1.0 / rate_limit)
cursor.close()
conn.close()
def lookup_mac_verbose(mac):
conn = get_connection()
cursor = conn.cursor(dictionary=True)
output = []
prefix = mac.lower().replace(":", "").replace("-", "")[:6]
output.append(f"🔍 Searching local database for prefix: {prefix}...")
cursor.execute("SELECT vendor_name, status FROM mac_vendors WHERE mac_prefix = %s", (prefix,))
row = cursor.fetchone()
if row:
output.append(f"✅ Found locally: {row['vendor_name']} (status: {row['status']})")
cursor.close()
conn.close()
return "\n".join(output)
output.append("❌ Not found locally.")
output.append("🌐 Querying API...")
url_template = current_app.config.get("OUI_API_URL", "https://api.maclookup.app/v2/macs/{}")
api_key = current_app.config.get("OUI_API_KEY", "")
headers = {"Authorization": f"Bearer {api_key}"} if api_key else {}
try:
url = url_template.format(prefix)
response = requests.get(url, headers=headers)
if response.status_code == 200:
data = response.json()
vendor_name = data.get("company", "").strip()
if vendor_name:
output.append(f"✅ Found via API: {vendor_name}")
output.append("💾 Inserting into local database...")
cursor.execute("""
INSERT INTO mac_vendors (mac_prefix, vendor_name, status, last_checked, last_updated)
VALUES (%s, %s, 'found', NOW(), NOW())
""", (prefix, vendor_name))
conn.commit()
else:
output.append("⚠️ API responded but no vendor name found. Not inserting.")
elif response.status_code == 404:
output.append("❌ Not found via API (404). Not inserting.")
else:
output.append(f"❌ API returned unexpected status: {response.status_code}")
except Exception as e:
output.append(f"🚨 Exception during API request: {e}")
cursor.close()
conn.close()
return "\n".join(output)