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

add console logging to radius

Browse Source
This commit is contained in:
Simon Cloutier
2025-04-09 08:51:14 -04:00
parent 0a254c9d20
commit ae4cd12f97
2 changed files with 64 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files

110
radius/main.py
View File

@@ -4,6 +4,7 @@ from pyrad.packet import AccessAccept, AccessReject
from datetime import datetime, timezone
import mysql.connector
import os
import traceback
DEFAULT_VLAN_ID = os.getenv("DEFAULT_VLAN", "505")
DENIED_VLAN = os.getenv("DENIED_VLAN", "999")
@@ -12,71 +13,86 @@ class MacRadiusServer(Server):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.db = mysql.connector.connect(
host=os.getenv('DB_HOST'),
port=int(os.getenv('DB_PORT', 3306)),
user=os.getenv('DB_USER'),
password=os.getenv('DB_PASSWORD'),
database=os.getenv('DB_NAME'),
)
try:
self.db = mysql.connector.connect(
host=os.getenv('DB_HOST'),
port=int(os.getenv('DB_PORT', 3306)),
user=os.getenv('DB_USER'),
password=os.getenv('DB_PASSWORD'),
database=os.getenv('DB_NAME'),
)
self.db.ping()
print("✅ Successfully connected to the database.")
except Exception as e:
print("❌ Failed to connect to the database.")
traceback.print_exc()
raise
def HandleAuthPacket(self, pkt):
username = pkt['User-Name'][0].upper()
cursor = self.db.cursor(dictionary=True)
now_utc = datetime.now(timezone.utc)
# Step 1: Check if the MAC exists in the users table
cursor.execute("SELECT vlan_id FROM users WHERE mac_address = %s", (username,))
result = cursor.fetchone()
print(f"\n📡 Received RADIUS Auth Request")
try:
username = pkt['User-Name'][0].upper()
print(f"→ Parsed MAC: {username}")
print(f"→ Attributes: {[f'{k}={v}' for k, v in pkt.items()]}")
reply = self.CreateReplyPacket(pkt)
cursor = self.db.cursor(dictionary=True)
now_utc = datetime.now(timezone.utc)
# Step 2: Handle the Access-Accept or Access-Reject scenario
if result:
vlan_id = result['vlan_id']
denied_vlan = os.getenv("DENIED_VLAN", "999")
cursor.execute("SELECT vlan_id FROM users WHERE mac_address = %s", (username,))
result = cursor.fetchone()
if vlan_id == denied_vlan:
reply.code = AccessReject
cursor.execute("""
INSERT INTO auth_logs (mac_address, reply, result, timestamp)
VALUES (%s, %s, %s, %s)
""", (username, "Access-Reject", f"Denied due to VLAN {denied_vlan}", now_utc))
self.db.commit()
print(f"[INFO] MAC {username} rejected due to VLAN {denied_vlan}")
reply = self.CreateReplyPacket(pkt)
if result:
vlan_id = result['vlan_id']
denied_vlan = os.getenv("DENIED_VLAN", "999")
if vlan_id == denied_vlan:
print(f"🚫 MAC {username} found, but on denied VLAN {vlan_id}")
reply.code = AccessReject
cursor.execute("""
INSERT INTO auth_logs (mac_address, reply, result, timestamp)
VALUES (%s, %s, %s, %s)
""", (username, "Access-Reject", f"Denied due to VLAN {denied_vlan}", now_utc))
self.db.commit()
else:
print(f"✅ MAC {username} found, assigning VLAN {vlan_id}")
reply.code = AccessAccept
reply.AddAttribute("Tunnel-Type", 13)
reply.AddAttribute("Tunnel-Medium-Type", 6)
reply.AddAttribute("Tunnel-Private-Group-Id", vlan_id)
cursor.execute("""
INSERT INTO auth_logs (mac_address, reply, result, timestamp)
VALUES (%s, %s, %s, %s)
""", (username, "Access-Accept", f"Assigned to VLAN {vlan_id}", now_utc))
self.db.commit()
else:
print(f"⚠️ MAC {username} not found, assigning fallback VLAN {DEFAULT_VLAN_ID}")
reply.code = AccessAccept
reply.AddAttribute("Tunnel-Type", 13)
reply.AddAttribute("Tunnel-Medium-Type", 6)
reply.AddAttribute("Tunnel-Private-Group-Id", vlan_id)
reply["Tunnel-Type"] = 13
reply["Tunnel-Medium-Type"] = 6
reply["Tunnel-Private-Group-Id"] = DEFAULT_VLAN_ID
cursor.execute("""
INSERT INTO auth_logs (mac_address, reply, result, timestamp)
VALUES (%s, %s, %s, %s)
""", (username, "Access-Accept", f"Assigned to VLAN {vlan_id}", now_utc))
""", (username, "Access-Accept", f"Assigned to fallback VLAN {DEFAULT_VLAN_ID}", now_utc))
self.db.commit()
print(f"[INFO] MAC {username} accepted and assigned to VLAN {vlan_id}")
else:
reply.code = AccessAccept
reply["Tunnel-Type"] = 13
reply["Tunnel-Medium-Type"] = 6
reply["Tunnel-Private-Group-Id"] = DEFAULT_VLAN_ID
self.SendReplyPacket(pkt.fd, reply)
print(f"📤 Response sent: {'Access-Accept' if reply.code == AccessAccept else 'Access-Reject'}\n")
cursor.execute("""
INSERT INTO auth_logs (mac_address, reply, result, timestamp)
VALUES (%s, %s, %s, %s)
""", (username, "Access-Accept", f"Assigned to fallback VLAN {DEFAULT_VLAN_ID}", now_utc))
self.db.commit()
print(f"[INFO] MAC {username} not found — assigned to fallback VLAN {DEFAULT_VLAN_ID}")
self.SendReplyPacket(pkt.fd, reply)
cursor.close()
except Exception as e:
print("❌ Error processing request:")
traceback.print_exc()
finally:
if 'cursor' in locals():
cursor.close()
if __name__ == '__main__':
print("🚀 Starting MacRadiusServer...")
srv = MacRadiusServer(dict=Dictionary("dictionary"))
srv.hosts["0.0.0.0"] = RemoteHost("0.0.0.0", os.getenv("RADIUS_SECRET", "testing123").encode(), "localhost")
print("📡 Listening on 0.0.0.0 for incoming RADIUS requests...")
srv.BindToAddress("0.0.0.0")
srv.Run()