113 lines
4.5 KiB
Python
113 lines
4.5 KiB
Python
import socket
|
|
import hashlib
|
|
import os
|
|
import logging
|
|
|
|
logger = logging.getLogger("uvicorn.error")
|
|
|
|
def authenticate_radius(username: str, password: str, server: str, secret: str, port: int = 1812, timeout: float = 3.0) -> bool:
|
|
"""
|
|
Performs RADIUS Access-Request authentication natively using raw UDP sockets.
|
|
Conforms to RFC 2865 standard for RADIUS protocol and PAP password encryption.
|
|
"""
|
|
if not server or not secret:
|
|
logger.error("RADIUS authentication failed: Server host or Shared Secret is not configured.")
|
|
return False
|
|
|
|
try:
|
|
secret_bytes = secret.encode('utf-8')
|
|
|
|
# Access-Request Header fields:
|
|
# Code: 1 (Access-Request)
|
|
# Identifier: 1 byte (random/sequential)
|
|
# Length: 2 bytes (20 + attributes length)
|
|
# Authenticator: 16 bytes (cryptographically strong random value)
|
|
identifier = os.urandom(1)[0]
|
|
authenticator = os.urandom(16)
|
|
|
|
# Attribute 1: User-Name (Type 1)
|
|
user_bytes = username.encode('utf-8')
|
|
attr_username = bytes([1, len(user_bytes) + 2]) + user_bytes
|
|
|
|
# Attribute 2: User-Password (Type 2, PAP Encryption)
|
|
# 1. Pad password with null bytes (\x00) to a multiple of 16 bytes
|
|
password_bytes = password.encode('utf-8')
|
|
pad_len = 16 - (len(password_bytes) % 16)
|
|
if pad_len == 16 and len(password_bytes) > 0:
|
|
pad_len = 0
|
|
if pad_len > 0:
|
|
password_bytes += b'\x00' * pad_len
|
|
|
|
# 2. Encrypt password chunks
|
|
# b(1) = MD5(Secret + Request Authenticator)
|
|
# c(1) = p(1) XOR b(1)
|
|
# b(2) = MD5(Secret + c(1))
|
|
# c(2) = p(2) XOR b(2)
|
|
# and so on...
|
|
encrypted_password = b''
|
|
last_chunk = authenticator
|
|
for i in range(0, len(password_bytes), 16):
|
|
chunk = password_bytes[i:i+16]
|
|
md5_hash = hashlib.md5(secret_bytes + last_chunk).digest()
|
|
encrypted_chunk = bytes(a ^ b for a, b in zip(chunk, md5_hash))
|
|
encrypted_password += encrypted_chunk
|
|
last_chunk = encrypted_chunk
|
|
|
|
attr_password = bytes([2, len(encrypted_password) + 2]) + encrypted_password
|
|
|
|
# Combine attributes
|
|
attributes = attr_username + attr_password
|
|
packet_len = 20 + len(attributes)
|
|
|
|
# Assemble complete packet
|
|
header = bytes([1, identifier, (packet_len >> 8) & 0xff, packet_len & 0xff]) + authenticator
|
|
packet = header + attributes
|
|
|
|
# UDP Send/Receive
|
|
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
|
sock.settimeout(timeout)
|
|
|
|
logger.info(f"Sending RADIUS Access-Request to {server}:{port} for user '{username}'...")
|
|
sock.sendto(packet, (server, port))
|
|
|
|
response, _ = sock.recvfrom(4096)
|
|
if len(response) < 20:
|
|
logger.warning("RADIUS response packet is too short.")
|
|
return False
|
|
|
|
resp_code = response[0]
|
|
resp_identifier = response[1]
|
|
resp_length = (response[2] << 8) + response[3]
|
|
resp_authenticator = response[4:20]
|
|
|
|
# Validate Identifier match
|
|
if resp_identifier != identifier:
|
|
logger.warning(f"RADIUS response identifier mismatch (sent {identifier}, received {resp_identifier}).")
|
|
return False
|
|
|
|
# Validate Response Authenticator:
|
|
# Response Authenticator = MD5(Code + ID + Length + Request Authenticator + Attributes + Secret)
|
|
resp_attributes = response[20:resp_length]
|
|
calc_auth = hashlib.md5(response[0:4] + authenticator + resp_attributes + secret_bytes).digest()
|
|
|
|
if calc_auth != resp_authenticator:
|
|
logger.warning("RADIUS response authenticator signature validation failed (spoofing check).")
|
|
return False
|
|
|
|
if resp_code == 2:
|
|
logger.info(f"RADIUS Access-Accept received for user '{username}'. Authentication successful!")
|
|
return True
|
|
elif resp_code == 3:
|
|
logger.warning(f"RADIUS Access-Reject received for user '{username}'.")
|
|
return False
|
|
else:
|
|
logger.warning(f"RADIUS server returned unknown response code: {resp_code}")
|
|
return False
|
|
|
|
except socket.timeout:
|
|
logger.error(f"RADIUS authentication timed out (server {server}:{port} unreachable).")
|
|
return False
|
|
except Exception as e:
|
|
logger.error(f"RADIUS authentication system exception: {e}")
|
|
return False
|