adif-api-reverse-engineering/adif_auth.py

#!/usr/bin/env python3
"""
ADIF API Authenticator - Réplica del Sistema Original

Este módulo es una réplica fiel del algoritmo de autenticación HMAC-SHA256
utilizado por la API de ADIF (El Cano Móvil), obtenido mediante ingeniería
reversa del código fuente original en ElcanoAuth.java.

El algoritmo sigue el patrón AWS Signature Version 4 con características
específicas de ADIF:
- Derivación de claves en cascada (date_key -> client_key -> signature_key)
- Orden NO alfabético de headers canónicos (crítico para el funcionamiento)
- Timestamp en formato ISO 8601 con zona horaria UTC

Fuente Original:
    apk_decompiled/sources/com/adif/elcanomovil/serviceNetworking/interceptors/auth/ElcanoAuth.java

Uso:
    auth = AdifAuthenticator(access_key="YOUR_KEY", secret_key="YOUR_KEY")
    headers = auth.get_auth_headers("POST", url, payload={...})
    response = requests.post(url, json=payload, headers=headers)
"""

import hashlib
import hmac
from datetime import datetime
import json
import uuid
from urllib.parse import urlparse


class AdifAuthenticator:
    """
    Implementa el algoritmo de autenticación HMAC-SHA256 de ADIF
    Similar a AWS Signature Version 4
    """

    # User-keys estáticas (diferentes de las claves HMAC)
    USER_KEY_CIRCULATION = "f4ce9fbfa9d721e39b8984805901b5df"
    USER_KEY_STATIONS = "0d021447a2fd2ac64553674d5a0c1a6f"

    def __init__(self, access_key, secret_key):
        """
        Inicializa el autenticador con las claves HMAC

        Args:
            access_key (str): Access key extraída de libapi-keys.so
            secret_key (str): Secret key extraída de libapi-keys.so
        """
        self.access_key = access_key
        self.secret_key = secret_key

    def get_timestamp(self, date=None):
        """
        Genera timestamp en formato ISO 8601 compacto UTC

        Args:
            date (datetime): Fecha a formatear (por defecto: ahora)

        Returns:
            str: Timestamp en formato yyyyMMddTHHmmssZ

        Ejemplo:
            "20251204T204637Z"
        """
        if date is None:
            date = datetime.utcnow()
        return date.strftime('%Y%m%dT%H%M%SZ')

    def get_date(self, date=None):
        """
        Genera fecha en formato compacto

        Args:
            date (datetime): Fecha a formatear (por defecto: ahora)

        Returns:
            str: Fecha en formato yyyyMMdd

        Ejemplo:
            "20251204"
        """
        if date is None:
            date = datetime.utcnow()
        return date.strftime('%Y%m%d')

    def format_payload(self, payload):
        """
        Formatea el payload JSON eliminando espacios y saltos de línea
        (ElcanoAuth.java:86-91)

        Args:
            payload: Diccionario o string con el payload

        Returns:
            str: Payload formateado sin espacios

        Ejemplo:
            Input:  {"page": {"pageNumber": 0}}
            Output: {"page":{"pageNumber":0}}
        """
        if payload is None:
            return ""

        if isinstance(payload, dict):
            payload = json.dumps(payload, separators=(',', ':'))

        return payload.replace('\r', '').replace('\n', '').replace(' ', '')

    def sha256_hash(self, text):
        """
        Calcula SHA-256 hash en formato hexadecimal
        (ElcanoAuth.java:185-193)

        Args:
            text (str): Texto a hashear

        Returns:
            str: Hash SHA-256 en hexadecimal (64 caracteres)
        """
        return hashlib.sha256(text.encode('utf-8')).hexdigest()

    def hmac_sha256(self, key, data):
        """
        Calcula HMAC-SHA256
        (ElcanoAuth.java:117-127)

        Args:
            key: Clave (str o bytes)
            data (str): Datos a firmar

        Returns:
            bytes: Firma HMAC-SHA256 (32 bytes)
        """
        if isinstance(key, str):
            key = key.encode('utf-8')

        return hmac.new(key, data.encode('utf-8'), hashlib.sha256).digest()

    def get_signature_key(self, date_simple, client):
        """
        Genera la clave de firma mediante derivación en cascada
        (ElcanoAuth.java:109-111)

        Proceso:
            kDate = HMAC(secretKey, date)
            kClient = HMAC(kDate, client)
            kSigning = HMAC(kClient, "elcano_request")

        Args:
            date_simple (str): Fecha en formato yyyyMMdd
            client (str): Nombre del cliente (ej: "AndroidElcanoApp")

        Returns:
            bytes: Clave de firma derivada (32 bytes)
        """
        k_date = self.hmac_sha256(self.secret_key, date_simple)
        k_client = self.hmac_sha256(k_date, client)
        k_signing = self.hmac_sha256(k_client, "elcano_request")

        return k_signing

    def prepare_canonical_request(self, method, path, params, payload,
                                   content_type, host, client, timestamp, user_id):
        """
        Prepara la petición canónica para firma
        (ElcanoAuth.java:129-172)

        Estructura:
            <HTTPMethod>
            <Path>
            <QueryString>
            content-type:<ContentType>
            x-elcano-client:<Client>
            x-elcano-date:<Timestamp>
            x-elcano-host:<Host>
            x-elcano-userid:<UserId>
            content-type;x-elcano-client;x-elcano-date;x-elcano-host;x-elcano-userid
            <SHA256HashOfPayload>

        Args:
            method (str): Método HTTP (GET, POST, etc.)
            path (str): Path de la URL
            params (str): Query string (puede ser vacío)
            payload: Body de la petición
            content_type (str): Content-Type
            host (str): Host del servidor
            client (str): Nombre del cliente
            timestamp (str): Timestamp de la petición
            user_id (str): UUID del usuario

        Returns:
            tuple: (canonical_request, signed_headers)
        """
        # Formatear payload
        formatted_payload = self.format_payload(payload)
        payload_hash = self.sha256_hash(formatted_payload)

        # Headers canónicos (ORDEN ESPECÍFICO, no alfabético completo!)
        # Nota: El orden DEBE coincidir exactamente con ElcanoAuth.java:137-165
        canonical_headers = (
            f"content-type:{content_type}\n"
            f"x-elcano-host:{host}\n"          # ← Segundo (antes de client!)
            f"x-elcano-client:{client}\n"      # ← Tercero
            f"x-elcano-date:{timestamp}\n"     # ← Cuarto
            f"x-elcano-userid:{user_id}\n"     # ← Quinto
        )

        # Lista de headers firmados (MISMO orden que canonical_headers)
        signed_headers = "content-type;x-elcano-host;x-elcano-client;x-elcano-date;x-elcano-userid"

        # Construir canonical request
        canonical_request = (
            f"{method}\n"
            f"{path}\n"
            f"{params}\n"
            f"{canonical_headers}"
            f"{signed_headers}\n"
            f"{payload_hash}"
        )

        return canonical_request, signed_headers

    def prepare_string_to_sign(self, timestamp, date_simple, client, user_id, canonical_request):
        """
        Prepara el string a firmar
        (ElcanoAuth.java:174-183)

        Estructura:
            HMAC-SHA256
            <Timestamp>
            <Date>/<Client>/<UserId>/elcano_request
            <SHA256HashOfCanonicalRequest>

        Args:
            timestamp (str): Timestamp ISO compacto
            date_simple (str): Fecha simple (yyyyMMdd)
            client (str): Nombre del cliente
            user_id (str): UUID del usuario
            canonical_request (str): Petición canónica

        Returns:
            str: String to sign
        """
        canonical_hash = self.sha256_hash(canonical_request)

        string_to_sign = (
            f"HMAC-SHA256\n"
            f"{timestamp}\n"
            f"{date_simple}/{client}/{user_id}/elcano_request\n"
            f"{canonical_hash}"
        )

        return string_to_sign

    def calculate_signature(self, string_to_sign, date_simple, client):
        """
        Calcula la firma final
        (ElcanoAuth.java:78-84)

        Args:
            string_to_sign (str): String preparado para firma
            date_simple (str): Fecha simple
            client (str): Nombre del cliente

        Returns:
            str: Firma en hexadecimal
        """
        signing_key = self.get_signature_key(date_simple, client)
        signature_bytes = hmac.new(signing_key, string_to_sign.encode('utf-8'), hashlib.sha256).digest()

        # Convertir a hexadecimal (minúsculas)
        signature = signature_bytes.hex()

        return signature

    def build_authorization_header(self, signature, date_simple, client, user_id, signed_headers):
        """
        Construye el header Authorization
        (ElcanoAuth.java:61-63)

        Formato:
            HMAC-SHA256 Credential=<accessKey>/<date>/<client>/<userId>/elcano_request,
            SignedHeaders=<headers>,Signature=<signature>

        Args:
            signature (str): Firma calculada
            date_simple (str): Fecha simple
            client (str): Nombre del cliente
            user_id (str): UUID del usuario
            signed_headers (str): Lista de headers firmados

        Returns:
            str: Header Authorization completo
        """
        return (
            f"HMAC-SHA256 "
            f"Credential={self.access_key}/{date_simple}/{client}/{user_id}/elcano_request,"
            f"SignedHeaders={signed_headers},"
            f"Signature={signature}"
        )

    def get_auth_headers(self, method, url, payload=None, user_id=None, date=None):
        """
        Genera todos los headers necesarios para autenticación

        Args:
            method (str): Método HTTP (GET, POST, etc.)
            url (str): URL completa de la petición
            payload: Body de la petición (dict o None)
            user_id (str): UUID del usuario (se genera si no se provee)
            date (datetime): Fecha de la petición (por defecto: ahora)

        Returns:
            dict: Headers completos para la petición

        Ejemplo:
            >>> auth = AdifAuthenticator(access_key="...", secret_key="...")
            >>> headers = auth.get_auth_headers(
            ...     "POST",
            ...     "https://circulacion.api.adif.es/path",
            ...     payload={"page": {"pageNumber": 0}}
            ... )
            >>> headers
            {
                "Content-Type": "application/json;charset=utf-8",
                "X-Elcano-Host": "circulacion.api.adif.es",
                "X-Elcano-Client": "AndroidElcanoApp",
                "X-Elcano-Date": "20251204T204637Z",
                "X-Elcano-UserId": "a1b2c3d4-...",
                "Authorization": "HMAC-SHA256 Credential=..."
            }
        """
        # Parse URL
        parsed = urlparse(url)
        host = parsed.netloc
        path = parsed.path
        params = parsed.query or ""

        # Defaults
        if user_id is None:
            user_id = str(uuid.uuid4())

        if date is None:
            date = datetime.utcnow()

        client = "AndroidElcanoApp"
        content_type = "application/json;charset=utf-8"

        # Generar timestamps
        timestamp = self.get_timestamp(date)
        date_simple = self.get_date(date)

        # 1. Preparar canonical request
        canonical_request, signed_headers = self.prepare_canonical_request(
            method, path, params, payload, content_type, host, client, timestamp, user_id
        )

        # 2. Preparar string to sign
        string_to_sign = self.prepare_string_to_sign(
            timestamp, date_simple, client, user_id, canonical_request
        )

        # 3. Calcular firma
        signature = self.calculate_signature(string_to_sign, date_simple, client)

        # 4. Construir header Authorization
        authorization = self.build_authorization_header(
            signature, date_simple, client, user_id, signed_headers
        )

        # 5. Retornar todos los headers
        return {
            "Content-Type": content_type,
            "X-Elcano-Host": host,
            "X-Elcano-Client": client,
            "X-Elcano-Date": timestamp,
            "X-Elcano-UserId": user_id,
            "Authorization": authorization
        }

    def get_user_key_for_url(self, url):
        """
        Obtiene la User-key estática correcta según la URL

        Args:
            url (str): URL de la petición

        Returns:
            str: User-key correspondiente
        """
        if "circulacion.api.adif.es" in url:
            return self.USER_KEY_CIRCULATION
        elif "estaciones.api.adif.es" in url:
            return self.USER_KEY_STATIONS
        else:
            return self.USER_KEY_CIRCULATION  # Por defecto


def example_usage():
    """
    Ejemplo de uso del autenticador
    """
    print("="*70)
    print("ADIF API Authenticator - Ejemplo de Uso")
    print("="*70)

    # PASO 1: Obtener las claves de libapi-keys.so
    # (Usar Ghidra o Frida para extraerlas)
    print("\n⚠️  IMPORTANTE: Reemplazar con las claves reales extraídas de libapi-keys.so")
    print("    Ver AUTHENTICATION_ALGORITHM.md para instrucciones de extracción\n")

    ACCESS_KEY = "and20210615"  # ✅ Extraído con Ghidra
    SECRET_KEY = "Jthjtr946RTt"  # ✅ Extraído con Ghidra

    # PASO 2: Crear el autenticador
    auth = AdifAuthenticator(access_key=ACCESS_KEY, secret_key=SECRET_KEY)

    # PASO 3: Preparar la petición
    url = "https://circulacion.api.adif.es/portroyalmanager/secure/circulationpaths/departures/traffictype/"
    payload = {
        "commercialService": "BOTH",
        "commercialStopType": "BOTH",
        "page": {"pageNumber": 0},
        "stationCode": "10200",  # Madrid Atocha
        "trafficType": "ALL"
    }

    # PASO 4: Generar headers de autenticación
    headers = auth.get_auth_headers("POST", url, payload=payload)

    # PASO 5: Añadir User-key estática
    headers["User-key"] = auth.get_user_key_for_url(url)

    # PASO 6: Mostrar resultado
    print("Headers generados:")
    print("-" * 70)
    for key, value in headers.items():
        print(f"{key}: {value}")

    print("\n" + "="*70)
    print("Para hacer la petición:")
    print("="*70)
    print("""
import requests

response = requests.post(
    url,
    json=payload,
    headers=headers
)

print(f"Status: {response.status_code}")
print(response.json())
    """)


if __name__ == "__main__":
    example_usage()