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