Nikolay Tatarinov
37218a436a
- Introduced LOOPBACK_HTTP_HOSTS in config.py to define valid loopback addresses for health-check URL and MINI_APP_SKIP_AUTH safety. - Updated run.py to utilize LOOPBACK_HTTP_HOSTS for determining the host in health check and authentication logic. - Enhanced test_app.py to skip tests if the required webapp output directory is not built, improving test reliability.
295 lines
10 KiB
Python
295 lines
10 KiB
Python
"""Load configuration from environment (e.g. .env via python-dotenv).
|
|
|
|
BOT_TOKEN is not validated on import; call require_bot_token() in the entry point
|
|
when running the bot. Numeric env vars (HTTP_PORT, INIT_DATA_MAX_AGE_SECONDS) use
|
|
safe parsing with defaults on invalid values.
|
|
"""
|
|
|
|
import logging
|
|
import os
|
|
import re
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
from dotenv import load_dotenv
|
|
|
|
from duty_teller.i18n.lang import normalize_lang
|
|
|
|
load_dotenv()
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Valid port range for HTTP_PORT.
|
|
HTTP_PORT_MIN, HTTP_PORT_MAX = 1, 65535
|
|
|
|
# Host values treated as loopback (for health-check URL and MINI_APP_SKIP_AUTH safety).
|
|
LOOPBACK_HTTP_HOSTS = ("127.0.0.1", "localhost", "::1", "")
|
|
|
|
# Project root (parent of duty_teller package). Used for webapp path, etc.
|
|
PROJECT_ROOT = Path(__file__).resolve().parent.parent
|
|
|
|
# Only digits for phone comparison (same when saving and when checking allowlist).
|
|
_PHONE_DIGITS_RE = re.compile(r"\D")
|
|
|
|
|
|
def normalize_phone(phone: str | None) -> str:
|
|
"""Return phone as digits only (spaces, +, parentheses, dashes removed).
|
|
|
|
Args:
|
|
phone: Raw phone string or None.
|
|
|
|
Returns:
|
|
Digits-only string, or empty string if None or empty.
|
|
"""
|
|
if not phone or not isinstance(phone, str):
|
|
return ""
|
|
return _PHONE_DIGITS_RE.sub("", phone.strip())
|
|
|
|
|
|
def _parse_phone_list(raw: str) -> set[str]:
|
|
"""Parse comma-separated phones into set of normalized (digits-only) strings."""
|
|
result = set()
|
|
for s in (raw or "").strip().split(","):
|
|
normalized = normalize_phone(s)
|
|
if normalized:
|
|
result.add(normalized)
|
|
return result
|
|
|
|
|
|
def _normalize_log_level(raw: str) -> str:
|
|
"""Return a valid log level name (DEBUG, INFO, WARNING, ERROR); default INFO."""
|
|
level = (raw or "").strip().upper()
|
|
if level in ("DEBUG", "INFO", "WARNING", "ERROR"):
|
|
return level
|
|
return "INFO"
|
|
|
|
|
|
def _parse_int_env(
|
|
name: str, default: int, min_val: int | None = None, max_val: int | None = None
|
|
) -> int:
|
|
"""Parse an integer from os.environ; use default on invalid or out-of-range. Log on fallback."""
|
|
raw = os.getenv(name)
|
|
if raw is None or raw == "":
|
|
return default
|
|
try:
|
|
value = int(raw.strip())
|
|
except ValueError:
|
|
logger.warning(
|
|
"Invalid %s=%r (expected integer); using default %s",
|
|
name,
|
|
raw,
|
|
default,
|
|
)
|
|
return default
|
|
if min_val is not None and value < min_val:
|
|
logger.warning(
|
|
"%s=%s is below minimum %s; using %s", name, value, min_val, min_val
|
|
)
|
|
return min_val
|
|
if max_val is not None and value > max_val:
|
|
logger.warning(
|
|
"%s=%s is above maximum %s; using %s", name, value, max_val, max_val
|
|
)
|
|
return max_val
|
|
return value
|
|
|
|
|
|
def _validate_database_url(url: str) -> bool:
|
|
"""Return True if URL looks like a supported SQLAlchemy URL (sqlite or postgres)."""
|
|
if not url or not isinstance(url, str):
|
|
return False
|
|
u = url.strip().split("?", 1)[0].lower()
|
|
return (
|
|
u.startswith("sqlite://")
|
|
or u.startswith("postgresql://")
|
|
or u.startswith("postgres://")
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Settings:
|
|
"""Injectable settings built from environment. Used in tests or when env is overridden."""
|
|
|
|
bot_token: str
|
|
database_url: str
|
|
bot_username: str
|
|
mini_app_base_url: str
|
|
mini_app_short_name: str
|
|
http_host: str
|
|
http_port: int
|
|
allowed_usernames: set[str]
|
|
admin_usernames: set[str]
|
|
allowed_phones: set[str]
|
|
admin_phones: set[str]
|
|
mini_app_skip_auth: bool
|
|
init_data_max_age_seconds: int
|
|
cors_origins: list[str]
|
|
external_calendar_ics_url: str
|
|
duty_display_tz: str
|
|
default_language: str
|
|
duty_pin_notify: bool
|
|
log_level: str
|
|
|
|
@classmethod
|
|
def from_env(cls) -> "Settings":
|
|
"""Build Settings from current environment (same logic as module-level variables).
|
|
|
|
Returns:
|
|
Settings instance with all fields populated from env.
|
|
"""
|
|
bot_token = os.getenv("BOT_TOKEN") or ""
|
|
raw_allowed = os.getenv("ALLOWED_USERNAMES", "").strip()
|
|
allowed = {
|
|
s.strip().lstrip("@").lower() for s in raw_allowed.split(",") if s.strip()
|
|
}
|
|
raw_admin = os.getenv("ADMIN_USERNAMES", "").strip()
|
|
admin = {
|
|
s.strip().lstrip("@").lower() for s in raw_admin.split(",") if s.strip()
|
|
}
|
|
allowed_phones = _parse_phone_list(os.getenv("ALLOWED_PHONES", ""))
|
|
admin_phones = _parse_phone_list(os.getenv("ADMIN_PHONES", ""))
|
|
raw_cors = os.getenv("CORS_ORIGINS", "").strip()
|
|
cors = (
|
|
[_o.strip() for _o in raw_cors.split(",") if _o.strip()]
|
|
if raw_cors and raw_cors != "*"
|
|
else ["*"]
|
|
)
|
|
raw_host = (os.getenv("HTTP_HOST") or "127.0.0.1").strip()
|
|
http_host = raw_host if raw_host else "127.0.0.1"
|
|
bot_username = (os.getenv("BOT_USERNAME", "") or "").strip().lstrip("@").lower()
|
|
database_url = os.getenv("DATABASE_URL", "sqlite:///data/duty_teller.db")
|
|
if not _validate_database_url(database_url):
|
|
logger.warning(
|
|
"DATABASE_URL does not look like a supported URL (sqlite:// or postgresql://); "
|
|
"DB connection may fail."
|
|
)
|
|
http_port = _parse_int_env(
|
|
"HTTP_PORT", 8080, min_val=HTTP_PORT_MIN, max_val=HTTP_PORT_MAX
|
|
)
|
|
init_data_max_age = _parse_int_env("INIT_DATA_MAX_AGE_SECONDS", 0, min_val=0)
|
|
return cls(
|
|
bot_token=bot_token,
|
|
database_url=database_url,
|
|
bot_username=bot_username,
|
|
mini_app_base_url=os.getenv("MINI_APP_BASE_URL", "").rstrip("/"),
|
|
mini_app_short_name=(os.getenv("MINI_APP_SHORT_NAME", "") or "").strip().strip("/"),
|
|
http_host=http_host,
|
|
http_port=http_port,
|
|
allowed_usernames=allowed,
|
|
admin_usernames=admin,
|
|
allowed_phones=allowed_phones,
|
|
admin_phones=admin_phones,
|
|
mini_app_skip_auth=os.getenv("MINI_APP_SKIP_AUTH", "").strip()
|
|
in ("1", "true", "yes"),
|
|
init_data_max_age_seconds=init_data_max_age,
|
|
cors_origins=cors,
|
|
external_calendar_ics_url=os.getenv(
|
|
"EXTERNAL_CALENDAR_ICS_URL", ""
|
|
).strip(),
|
|
duty_display_tz=os.getenv("DUTY_DISPLAY_TZ", "Europe/Moscow").strip()
|
|
or "Europe/Moscow",
|
|
default_language=normalize_lang(os.getenv("DEFAULT_LANGUAGE", "en")),
|
|
duty_pin_notify=os.getenv("DUTY_PIN_NOTIFY", "1").strip().lower()
|
|
not in ("0", "false", "no"),
|
|
log_level=_normalize_log_level(os.getenv("LOG_LEVEL", "INFO")),
|
|
)
|
|
|
|
|
|
# Single source of truth: load once at import; entry point must check BOT_TOKEN when needed.
|
|
_settings = Settings.from_env()
|
|
|
|
BOT_TOKEN = _settings.bot_token
|
|
DATABASE_URL = _settings.database_url
|
|
BOT_USERNAME = _settings.bot_username
|
|
MINI_APP_BASE_URL = _settings.mini_app_base_url
|
|
MINI_APP_SHORT_NAME = _settings.mini_app_short_name
|
|
HTTP_HOST = _settings.http_host
|
|
HTTP_PORT = _settings.http_port
|
|
ALLOWED_USERNAMES = _settings.allowed_usernames
|
|
ADMIN_USERNAMES = _settings.admin_usernames
|
|
ALLOWED_PHONES = _settings.allowed_phones
|
|
ADMIN_PHONES = _settings.admin_phones
|
|
MINI_APP_SKIP_AUTH = _settings.mini_app_skip_auth
|
|
INIT_DATA_MAX_AGE_SECONDS = _settings.init_data_max_age_seconds
|
|
CORS_ORIGINS = _settings.cors_origins
|
|
EXTERNAL_CALENDAR_ICS_URL = _settings.external_calendar_ics_url
|
|
DUTY_DISPLAY_TZ = _settings.duty_display_tz
|
|
DEFAULT_LANGUAGE = _settings.default_language
|
|
DUTY_PIN_NOTIFY = _settings.duty_pin_notify
|
|
LOG_LEVEL = getattr(logging, _settings.log_level.upper(), logging.INFO)
|
|
LOG_LEVEL_STR = _settings.log_level
|
|
|
|
|
|
def is_admin(username: str) -> bool:
|
|
"""Check if Telegram username is in ADMIN_USERNAMES.
|
|
|
|
Args:
|
|
username: Telegram username (with or without @; case-insensitive).
|
|
|
|
Returns:
|
|
True if in ADMIN_USERNAMES.
|
|
"""
|
|
return (username or "").strip().lower() in ADMIN_USERNAMES
|
|
|
|
|
|
def can_access_miniapp(username: str) -> bool:
|
|
"""Check if username is allowed to open the calendar Miniapp (env allowlist only).
|
|
|
|
Legacy: Miniapp access in production is determined by repository.can_access_miniapp_for_telegram_user
|
|
(DB roles + fallback to ADMIN_*). This function is kept for tests and backward compatibility.
|
|
|
|
Args:
|
|
username: Telegram username (with or without @; case-insensitive).
|
|
|
|
Returns:
|
|
True if in ALLOWED_USERNAMES or ADMIN_USERNAMES.
|
|
"""
|
|
u = (username or "").strip().lower()
|
|
return u in ALLOWED_USERNAMES or u in ADMIN_USERNAMES
|
|
|
|
|
|
def can_access_miniapp_by_phone(phone: str | None) -> bool:
|
|
"""Check if phone (set via /set_phone) is allowed to open the Miniapp (env allowlist only).
|
|
|
|
Legacy: Miniapp access in production is determined by repository.can_access_miniapp_for_telegram_user
|
|
(DB roles + fallback to ADMIN_*). This function is kept for tests and backward compatibility.
|
|
|
|
Args:
|
|
phone: Raw phone string or None.
|
|
|
|
Returns:
|
|
True if normalized phone is in ALLOWED_PHONES or ADMIN_PHONES.
|
|
"""
|
|
normalized = normalize_phone(phone)
|
|
if not normalized:
|
|
return False
|
|
return normalized in ALLOWED_PHONES or normalized in ADMIN_PHONES
|
|
|
|
|
|
def is_admin_by_phone(phone: str | None) -> bool:
|
|
"""Check if phone is in ADMIN_PHONES.
|
|
|
|
Args:
|
|
phone: Raw phone string or None.
|
|
|
|
Returns:
|
|
True if normalized phone is in ADMIN_PHONES.
|
|
"""
|
|
normalized = normalize_phone(phone)
|
|
return bool(normalized and normalized in ADMIN_PHONES)
|
|
|
|
|
|
def require_bot_token() -> None:
|
|
"""Raise SystemExit with a clear message if BOT_TOKEN is not set.
|
|
|
|
Call from the application entry point (e.g. main.py or duty_teller.run) so the
|
|
process exits with a helpful message instead of failing later.
|
|
|
|
Raises:
|
|
SystemExit: If BOT_TOKEN is empty.
|
|
"""
|
|
if not BOT_TOKEN:
|
|
raise SystemExit(
|
|
"BOT_TOKEN is not set. Copy .env.example to .env and set your token from @BotFather."
|
|
)
|