lserp_cs_6.0/插件库/astrbot_plugin_lserp/session_auth.py

from __future__ import annotations

import hashlib
import hmac
import re


SAFE_CONVERSATION_ID = re.compile(r"^[A-Za-z0-9_.:-]{8,128}$")
PROCESS_BOUND_SESSION_ID = re.compile(
    r"(?:^|-)p(?P<pid>[1-9][0-9]{0,9})"
    r"-s(?P<started>[0-9]{9,12})"
    r"-c(?P<scope>[a-f0-9]{32})-[A-Fa-f0-9]{32}$"
)
PROCESS_BOUND_SESSION_SUFFIX = re.compile(
    r"^p(?P<pid>[1-9][0-9]{0,9})"
    r"-s(?P<started>[0-9]{9,12})"
    r"-c(?P<scope>[a-f0-9]{32})-[A-Fa-f0-9]{32}$"
)
SAFE_DATABASE_SCOPE = re.compile(r"^[a-f0-9]{64}$")
SAFE_SESSION_SCOPE_TOKEN = re.compile(r"^[a-f0-9]{32}$")
SESSION_SCOPE_DOMAIN = "lserp-pet-session-scope-v3\n"


def conversation_id(session_id: str | None) -> str:
    value = (session_id or "").strip()
    if "!" in value:
        return value.rsplit("!", 1)[-1]
    return value


def bridge_client_session_id(session_id: str | None) -> str:
    value = conversation_id(session_id)
    if SAFE_CONVERSATION_ID.fullmatch(value) is None:
        raise ValueError("AstrBot conversation ID 格式无效。")
    digest = hashlib.sha256(value.encode("utf-8")).hexdigest()[:32]
    return "astrbot-" + digest


def bridge_process_id(session_id: str | None) -> int | None:
    value = conversation_id(session_id)
    if SAFE_CONVERSATION_ID.fullmatch(value) is None:
        raise ValueError("AstrBot conversation ID 格式无效。")
    match = PROCESS_BOUND_SESSION_ID.search(value)
    if match is None:
        return None
    process_id = int(match.group("pid"))
    if process_id > 2_147_483_647:
        raise ValueError("ERP process ID 格式无效。")
    return process_id


def bridge_process_started_at_unix_seconds(session_id: str | None) -> int | None:
    value = conversation_id(session_id)
    if SAFE_CONVERSATION_ID.fullmatch(value) is None:
        raise ValueError("AstrBot conversation ID 格式无效。")
    match = PROCESS_BOUND_SESSION_ID.search(value)
    if match is None or match.group("started") is None:
        return None
    started_at = int(match.group("started"))
    if started_at <= 0 or started_at > 253_402_300_799:
        raise ValueError("ERP process start fingerprint 格式无效。")
    return started_at


def bridge_session_scope_token(session_id: str | None) -> str | None:
    value = conversation_id(session_id)
    if SAFE_CONVERSATION_ID.fullmatch(value) is None:
        raise ValueError("AstrBot conversation ID 格式无效。")
    match = PROCESS_BOUND_SESSION_ID.search(value)
    if match is None or match.group("scope") is None:
        return None
    token = match.group("scope")
    if SAFE_SESSION_SCOPE_TOKEN.fullmatch(token) is None:
        raise ValueError("ERP session scope token 格式无效。")
    return token


def compute_session_scope_token(
    database_scope_fingerprint: str,
    user_id: str,
    user_name: str,
    account_book: str,
    sub_system_id: str,
    is_administrator: bool,
) -> str:
    database = _scope_text(
        database_scope_fingerprint,
        "databaseScopeFingerprint",
    ).lower()
    if SAFE_DATABASE_SCOPE.fullmatch(database) is None:
        raise ValueError("ERP database scope fingerprint 格式无效。")
    values = (
        ("databaseScopeFingerprint", database),
        ("userId", _scope_text(user_id, "userId")),
        ("userName", _scope_text(user_name, "userName")),
        ("accountBook", _scope_text(account_book, "accountBook")),
        ("subSystemId", _scope_text(sub_system_id, "subSystemId")),
        (
            "isAdministrator",
            "true" if _scope_bool(is_administrator) else "false",
        ),
    )
    canonical = SESSION_SCOPE_DOMAIN + "".join(
        f"{name}={len(value.encode('utf-8'))}:{value}\n"
        for name, value in values
    )
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()[:32]


def session_scope_matches(
    expected_token: str,
    context: dict[str, object],
) -> bool:
    if SAFE_SESSION_SCOPE_TOKEN.fullmatch(expected_token or "") is None:
        return False
    try:
        actual = compute_session_scope_token(
            context["databaseScopeFingerprint"],
            context["userId"],
            context["userName"],
            context["accountBook"],
            context["subSystemId"],
            context["isAdministrator"],
        )
    except (KeyError, TypeError, ValueError):
        return False
    return hmac.compare_digest(expected_token, actual)


def _scope_bool(value: object) -> bool:
    if type(value) is not bool:
        raise ValueError("ERP session administrator scope 格式无效。")
    return value


def _scope_text(value: object, name: str) -> str:
    if (
        not isinstance(value, str)
        or not value.strip()
        or value != value.strip()
        or len(value) > 256
        or any(ord(character) < 32 or ord(character) == 127 for character in value)
    ):
        raise ValueError(f"ERP session scope field 格式无效：{name}。")
    return value


def is_authorized_session(
    session_id: str | None,
    prefix: str,
    platform_name: str | None = "webchat",
) -> bool:
    normalized_prefix = (prefix or "").strip()
    identifier = conversation_id(session_id)
    return (
        3 <= len(normalized_prefix) <= 64
        and normalized_prefix.endswith("-")
        and SAFE_CONVERSATION_ID.fullmatch(normalized_prefix + "bound") is not None
        and (platform_name or "").lower() == "webchat"
        and SAFE_CONVERSATION_ID.fullmatch(identifier) is not None
        and identifier.startswith(normalized_prefix)
        and PROCESS_BOUND_SESSION_SUFFIX.fullmatch(
            identifier[len(normalized_prefix):]
        )
        is not None
    )