feat: implementar pestaña de Gestion de Cuentas y autenticacion global RADIUS con JIT provisioning

tests/test_auth_radius.py

@@ -0,0 +1,140 @@
+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()