141 lines
5.9 KiB
Python
141 lines
5.9 KiB
Python
import socket
|
|
import threading
|
|
import pytest
|
|
from datetime import datetime, timezone
|
|
import hashlib
|
|
from app.core.radius import authenticate_radius
|
|
from app.core.database import AsyncSessionLocal, init_db
|
|
from app.models.models import User, SystemSetting
|
|
from app.api.auth import login
|
|
from app.schemas.schemas import LoginRequest
|
|
from sqlalchemy import select
|
|
|
|
class MockRadiusServer:
|
|
def __init__(self, host="127.0.0.1", port=18120, secret="my-secret"):
|
|
self.host = host
|
|
self.port = port
|
|
self.secret = secret
|
|
self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
|
self.sock.bind((self.host, self.port))
|
|
self.running = False
|
|
|
|
def start(self):
|
|
self.running = True
|
|
self.thread = threading.Thread(target=self._run, daemon=True)
|
|
self.thread.start()
|
|
|
|
def stop(self):
|
|
self.running = False
|
|
self.sock.close()
|
|
|
|
def _run(self):
|
|
while self.running:
|
|
try:
|
|
data, addr = self.sock.recvfrom(4096)
|
|
if len(data) < 20:
|
|
continue
|
|
code = data[0]
|
|
identifier = data[1]
|
|
length = (data[2] << 8) + data[3]
|
|
req_authenticator = data[4:20]
|
|
attributes = data[20:length]
|
|
|
|
# Parse Attributes
|
|
idx = 0
|
|
username = ""
|
|
password = ""
|
|
while idx < len(attributes):
|
|
attr_type = attributes[idx]
|
|
attr_len = attributes[idx+1]
|
|
attr_val = attributes[idx+2 : idx+attr_len]
|
|
if attr_type == 1:
|
|
username = attr_val.decode('utf-8')
|
|
elif attr_type == 2:
|
|
# Decrypt PAP password
|
|
decrypted = b''
|
|
last_chunk = req_authenticator
|
|
secret_bytes = self.secret.encode('utf-8')
|
|
for k in range(0, len(attr_val), 16):
|
|
chunk = attr_val[k:k+16]
|
|
md5_hash = hashlib.md5(secret_bytes + last_chunk).digest()
|
|
dec_chunk = bytes(a ^ b for a, b in zip(chunk, md5_hash))
|
|
decrypted += dec_chunk
|
|
last_chunk = chunk
|
|
password = decrypted.decode('utf-8').rstrip('\x00')
|
|
idx += attr_len
|
|
|
|
# Verify password. Let's make "radiuspass" the valid password
|
|
response_code = 3 # Access-Reject
|
|
if username == "radiususer" and password == "radiuspass":
|
|
response_code = 2 # Access-Accept
|
|
|
|
# Build response packet
|
|
# Response Authenticator = MD5(Code + ID + Length + Request Authenticator + Attributes + Secret)
|
|
resp_length = 20
|
|
header = bytes([response_code, identifier, (resp_length >> 8) & 0xff, resp_length & 0xff])
|
|
resp_authenticator = hashlib.md5(header + req_authenticator + self.secret.encode('utf-8')).digest()
|
|
|
|
response_packet = header[0:4] + resp_authenticator
|
|
self.sock.sendto(response_packet, addr)
|
|
except Exception:
|
|
break
|
|
|
|
def test_radius_authentication_success_and_fail():
|
|
server = MockRadiusServer(secret="testing-secret")
|
|
server.start()
|
|
try:
|
|
# 1. Access-Accept
|
|
ok = authenticate_radius("radiususer", "radiuspass", "127.0.0.1", "testing-secret", port=18120, timeout=1.0)
|
|
assert ok is True
|
|
|
|
# 2. Access-Reject (wrong password)
|
|
ok = authenticate_radius("radiususer", "wrongpass", "127.0.0.1", "testing-secret", port=18120, timeout=1.0)
|
|
assert ok is False
|
|
|
|
# 3. Access-Reject (wrong secret)
|
|
ok = authenticate_radius("radiususer", "radiuspass", "127.0.0.1", "bad-secret", port=18120, timeout=1.0)
|
|
assert ok is False
|
|
finally:
|
|
server.stop()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_radius_login_jit_provisioning():
|
|
"""
|
|
Tests RADIUS login flow: JIT creation of a local User when RADIUS Access-Accept succeeds.
|
|
"""
|
|
await init_db()
|
|
server = MockRadiusServer(secret="testing-secret")
|
|
server.start()
|
|
|
|
try:
|
|
async with AsyncSessionLocal() as db:
|
|
# Configure global settings to RADIUS authentication
|
|
db.add(SystemSetting(key="auth_mode", value="radius"))
|
|
db.add(SystemSetting(key="radius_host", value="127.0.0.1"))
|
|
db.add(SystemSetting(key="radius_port", value="18120"))
|
|
db.add(SystemSetting(key="radius_secret", value="testing-secret"))
|
|
await db.commit()
|
|
|
|
# Execute API login for 'radiususer@oneverdrive.local' with 'radiuspass'
|
|
login_req = LoginRequest(email="radiususer@oneverdrive.local", password="radiuspass")
|
|
res = await login(credentials=login_req, db=db)
|
|
|
|
assert res["access_token"] is not None
|
|
assert res["user"]["email"] == "radiususer@oneverdrive.local"
|
|
assert res["user"]["role"] == "OPERATOR"
|
|
assert res["user"]["auth_source"] == "radius"
|
|
|
|
# Check that user was saved to the SQLite DB
|
|
stmt = select(User).where(User.email == "radiususer@oneverdrive.local")
|
|
user_db = (await db.execute(stmt)).scalar_one_or_none()
|
|
assert user_db is not None
|
|
assert user_db.auth_source == "radius"
|
|
assert user_db.is_active is True
|
|
|
|
# Revert global settings by deleting test configuration keys
|
|
from sqlalchemy import delete
|
|
await db.execute(delete(SystemSetting).where(SystemSetting.key.in_(["auth_mode", "radius_host", "radius_port", "radius_secret"])))
|
|
await db.commit()
|
|
finally:
|
|
server.stop()
|