vpn-telegram-bot/services/env_file.py

"""Чтение/запись allowlist-ключей в .env (админка платежей)."""

from __future__ import annotations

import os
import re
from pathlib import Path

from dotenv import dotenv_values

from config import ROOT_DIR

# Ключи, которыми можно управлять из админки
PAYMENT_ENV_KEYS: tuple[str, ...] = (
    "CRYPTO_BOT_TOKEN",
    "CRYPTO_PAY_TESTNET",
    "HELEKET_MERCHANT_ID",
    "HELEKET_API_KEY",
    "HELEKET_CALLBACK_URL",
    "HELEKET_CURRENCY",
    "LAVA_SHOP_ID",
    "LAVA_SECRET_KEY",
    "LAVA_ADDITIONAL_KEY",
    "LAVA_HOOK_URL",
    "LAVA_SUCCESS_URL",
    "LAVA_FAIL_URL",
    "LAVA_EXPIRE_MINUTES",
    "DIGISELLER_SELLER_ID",
    "DIGISELLER_API_KEY",
    "DIGISELLER_PRODUCT_IDS",
    "DIGISELLER_PAY_URL",
    "DIGISELLER_CREDIT_FROM",
    "DIGISELLER_CALLBACK_URL",
    "ANYPAY_API_ID",
    "ANYPAY_API_KEY",
    "ANYPAY_PROJECT_ID",
    "ANYPAY_CURRENCY",
    "ANYPAY_METHOD",
    "ANYPAY_SUCCESS_URL",
    "ANYPAY_FAIL_URL",
    "ANYPAY_EMAIL",
    "ANYPAY_CHECK_IP",
    "STARS_ENABLED",
    "STARS_RUB_PER_STAR",
    "TOPUP_CODE_PAY_URL",
)

SECRET_ENV_KEYS = frozenset(
    {
        "CRYPTO_BOT_TOKEN",
        "HELEKET_API_KEY",
        "LAVA_SECRET_KEY",
        "LAVA_ADDITIONAL_KEY",
        "DIGISELLER_API_KEY",
        "ANYPAY_API_KEY",
        "SMTP_PASSWORD",
    }
)

BOOL_ENV_KEYS = frozenset(
    {"CRYPTO_PAY_TESTNET", "STARS_ENABLED", "ANYPAY_CHECK_IP", "SMTP_TLS", "SMTP_SSL"}
)

_KEY_LINE_RE = re.compile(
    r"^(?P<prefix>\s*(?:export\s+)?)(?P<key>[A-Za-z_][A-Za-z0-9_]*)\s*=\s*(?P<value>.*)$"
)


def env_file_path() -> Path:
    raw = (os.getenv("ENV_FILE") or "").strip()
    if raw:
        return Path(raw)
    return ROOT_DIR / ".env"


def env_writable(path: Path | None = None) -> bool:
    p = path or env_file_path()
    try:
        if p.exists():
            return os.access(p, os.W_OK)
        parent = p.parent
        return parent.exists() and os.access(parent, os.W_OK)
    except OSError:
        return False


def mask_secret(value: str) -> str:
    v = (value or "").strip()
    if not v:
        return ""
    if len(v) <= 6:
        return "••••••"
    return f"{v[:3]}…{v[-4:]}"


def read_env_values(path: Path | None = None) -> dict[str, str]:
    p = path or env_file_path()
    if not p.exists():
        return {}
    raw = dotenv_values(p)
    out: dict[str, str] = {}
    for k, v in raw.items():
        if k is None:
            continue
        out[str(k)] = "" if v is None else str(v)
    return out


def _format_env_value(value: str) -> str:
    """Значение для .env без лишних кавычек, если не нужны."""
    if value == "":
        return ""
    if any(ch in value for ch in ' \t#"\'\\$`\n\r'):
        escaped = value.replace("\\", "\\\\").replace('"', '\\"')
        return f'"{escaped}"'
    return value


def _write_env_inplace(path: Path, text: str) -> None:
    """Перезаписать содержимое файла без rename (Docker bind-mount .env)."""
    data = text.encode("utf-8")
    with open(path, "r+b") as fh:
        fh.truncate(0)
        fh.seek(0)
        fh.write(data)
        fh.flush()
        os.fsync(fh.fileno())


def upsert_env_keys(
    updates: dict[str, str | None],
    *,
    path: Path | None = None,
    allowlist: tuple[str, ...] = PAYMENT_ENV_KEYS,
) -> Path:
    """Записать ключи в .env и в os.environ. None / пустая строка удаляет ключ.

    Пишем in-place: docker volume mount файла не позволяет atomic rename
    (Errno 16 Device or resource busy).
    """
    p = path or env_file_path()
    allowed = frozenset(allowlist)
    filtered = {k: v for k, v in updates.items() if k in allowed}
    if not filtered:
        return p

    if not p.exists():
        p.parent.mkdir(parents=True, exist_ok=True)
        p.write_text("", encoding="utf-8")

    raw = p.read_text(encoding="utf-8")
    lines = raw.splitlines(keepends=True)
    if lines and not lines[-1].endswith(("\n", "\r")):
        # нормализуем конец файла при дальнейшей склейке
        pass

    pending = dict(filtered)
    new_lines: list[str] = []
    for line in lines:
        stripped = line.lstrip()
        if not stripped or stripped.startswith("#"):
            new_lines.append(line)
            continue
        m = _KEY_LINE_RE.match(line.rstrip("\r\n"))
        if not m:
            new_lines.append(line)
            continue
        key = m.group("key")
        if key not in pending:
            new_lines.append(line)
            continue
        value = pending.pop(key)
        ending = "\n" if line.endswith("\n") else ("\r\n" if line.endswith("\r\n") else "\n")
        if value is None or value == "":
            os.environ.pop(key, None)
            continue
        prefix = m.group("prefix") or ""
        new_lines.append(f"{prefix}{key}={_format_env_value(value)}{ending}")
        os.environ[key] = value

    for key, value in pending.items():
        if value is None or value == "":
            os.environ.pop(key, None)
            continue
        if new_lines and not new_lines[-1].endswith(("\n", "\r")):
            new_lines[-1] = new_lines[-1] + "\n"
        new_lines.append(f"{key}={_format_env_value(value)}\n")
        os.environ[key] = value

    text = "".join(new_lines)
    if text and not text.endswith("\n"):
        text += "\n"
    _write_env_inplace(p, text)
    return p