import hashlib import hmac import json import unittest from types import SimpleNamespace from unittest.mock import AsyncMock, patch from urllib.parse import parse_qs, urlsplit from aiohttp import web from bot.app.web import admin_api, subscription_webapp from bot.app.web.webapp_auth import ( create_telegram_oauth_nonce, create_webapp_session_token, verify_telegram_oauth_nonce, ) from bot.handlers.user.payment import yookassa_webhook_route from bot.services.crypto_pay_service import CryptoPayService from bot.services.freekassa_service import FreeKassaService from bot.utils.request_security import request_client_ip from config.settings import Settings from db.database_setup import redacted_database_url class RequestSecurityTests(unittest.IsolatedAsyncioTestCase): async def test_request_client_ip_uses_last_forwarded_for_value_for_trusted_proxy(self): request = SimpleNamespace( remote="127.0.0.1", headers={"X-Forwarded-For": "203.0.113.10, 198.51.100.7"}, ) self.assertEqual( request_client_ip(request, trusted_proxies=["127.0.0.1"]), "198.51.100.7", ) async def test_yookassa_webhook_rejects_untrusted_ip_before_reading_body(self): request = SimpleNamespace( app={ "bot": object(), "i18n": object(), "settings": SimpleNamespace(trusted_proxies=["127.0.0.1"]), "panel_service": object(), "subscription_service": object(), "referral_service": object(), "lknpd_service": None, "async_session_factory": object(), }, headers={}, remote="203.0.113.50", json=AsyncMock(side_effect=AssertionError("request.json() must not be called")), ) response = await yookassa_webhook_route(request) self.assertEqual(response.status, 403) request.json.assert_not_awaited() class FreeKassaServiceTests(unittest.TestCase): def _make_service(self) -> FreeKassaService: settings = SimpleNamespace( FREEKASSA_ENABLED=True, FREEKASSA_MERCHANT_ID="123456", FREEKASSA_API_KEY="api-key", FREEKASSA_SECOND_SECRET="second-secret", DEFAULT_CURRENCY_SYMBOL="RUB", FREEKASSA_PAYMENT_IP="203.0.113.10", FREEKASSA_PAYMENT_METHOD_ID=44, FREEKASSA_TRUSTED_IPS="127.0.0.1,203.0.113.0/24", trusted_proxies=["127.0.0.1"], freekassa_trusted_ips=["127.0.0.1", "203.0.113.0/24"], ) return FreeKassaService( bot=object(), settings=settings, i18n=object(), async_session_factory=object(), subscription_service=object(), referral_service=object(), ) def test_validate_signature_accepts_hmac_sha256_raw_body(self): service = self._make_service() raw_body = b'{"amount":"199.00","o":"42"}' expected_signature = hmac.new( service.second_secret.encode("utf-8"), raw_body, hashlib.sha256, ).hexdigest() self.assertTrue(service._validate_signature(raw_body, expected_signature)) def test_validate_signature_rejects_wrong_signature(self): service = self._make_service() self.assertFalse(service._validate_signature(b"payload", "not-a-signature")) def test_webhook_rejects_unauthorized_ip_before_body_read(self): service = self._make_service() request = SimpleNamespace( remote="198.51.100.250", headers={}, read=AsyncMock(side_effect=AssertionError("request.read() must not be called")), ) response = asyncio_run(service.webhook_route(request)) self.assertEqual(response.status, 403) request.read.assert_not_awaited() class CryptoPayServiceTests(unittest.TestCase): def _make_service(self) -> CryptoPayService: service = CryptoPayService.__new__(CryptoPayService) service.token = "cryptopay-token" return service def test_validate_webhook_signature_accepts_valid_signature(self): service = self._make_service() raw_body = b'{"payload":"42"}' expected_signature = hmac.new( hashlib.sha256(service.token.encode("utf-8")).digest(), raw_body, hashlib.sha256, ).hexdigest() self.assertTrue(service._validate_webhook_signature(raw_body, expected_signature)) def test_validate_webhook_signature_rejects_invalid_signature(self): service = self._make_service() self.assertFalse(service._validate_webhook_signature(b"payload", "not-a-signature")) class WebAppSecurityTests(unittest.IsolatedAsyncioTestCase): def test_auth_response_sets_cookies_and_does_not_return_session_token(self): settings = SimpleNamespace( WEBAPP_SESSION_SECRET="session-secret", WEBAPP_SESSION_TTL_SECONDS=3600, ) response = subscription_webapp._build_webapp_auth_response( settings, {"user_id": 321}, token="raw-session-token", csrf_token="csrf-token", ) payload = json.loads(response.text) self.assertTrue(payload["ok"]) self.assertEqual(payload["csrf_token"], "csrf-token") self.assertNotIn("token", payload) self.assertEqual(response.cookies["rw_webapp_session"].value, "raw-session-token") self.assertTrue(response.cookies["rw_webapp_session"]["httponly"]) self.assertEqual(response.cookies["rw_webapp_csrf"].value, "csrf-token") self.assertFalse(response.cookies["rw_webapp_csrf"]["httponly"]) def test_require_user_id_falls_back_to_cookie_session(self): settings = SimpleNamespace( WEBAPP_SESSION_SECRET="session-secret", WEBAPP_SESSION_TTL_SECONDS=3600, ) token = create_webapp_session_token(settings, 321) request = SimpleNamespace( app={"settings": settings}, headers={}, cookies={"rw_webapp_session": token}, ) self.assertEqual(subscription_webapp._require_user_id(request), 321) async def test_csrf_middleware_rejects_mismatched_token_when_cookie_session_exists(self): settings = SimpleNamespace( WEBAPP_SESSION_SECRET="session-secret", WEBAPP_SESSION_TTL_SECONDS=3600, ) request = SimpleNamespace( method="POST", path="/api/payments", headers={"X-CSRF-Token": "bad-token"}, cookies={"rw_webapp_session": "session-cookie", "rw_webapp_csrf": "good-token"}, app={"settings": settings}, ) handler = AsyncMock(return_value=web.Response(text="ok")) response = await subscription_webapp._csrf_protection_middleware(request, handler) self.assertEqual(response.status, 403) handler.assert_not_awaited() async def test_csrf_middleware_allows_matching_token_when_cookie_session_exists(self): settings = SimpleNamespace( WEBAPP_SESSION_SECRET="session-secret", WEBAPP_SESSION_TTL_SECONDS=3600, ) request = SimpleNamespace( method="POST", path="/api/payments", headers={"X-CSRF-Token": "good-token"}, cookies={"rw_webapp_session": "session-cookie", "rw_webapp_csrf": "good-token"}, app={"settings": settings}, ) handler = AsyncMock(return_value=web.Response(text="ok")) response = await subscription_webapp._csrf_protection_middleware(request, handler) self.assertEqual(response.text, "ok") handler.assert_awaited_once() async def test_csrf_middleware_allows_valid_bearer_authorization_for_compatibility(self): settings = SimpleNamespace( WEBAPP_SESSION_SECRET="session-secret", WEBAPP_SESSION_TTL_SECONDS=3600, ) token = create_webapp_session_token(settings, 321) request = SimpleNamespace( method="POST", path="/api/payments", headers={ "Authorization": f"Bearer {token}", "X-CSRF-Token": "bad-token", }, cookies={"rw_webapp_session": "session-cookie", "rw_webapp_csrf": "good-token"}, app={"settings": settings}, ) handler = AsyncMock(return_value=web.Response(text="ok")) response = await subscription_webapp._csrf_protection_middleware(request, handler) self.assertEqual(response.text, "ok") handler.assert_awaited_once() def test_email_payload_rejects_overlong_email(self): long_email = ("a" * 245) + "@example.com" model, response = subscription_webapp._validate_model_payload( subscription_webapp.WebAppEmailPayload, {"email": long_email}, ) self.assertIsNone(model) self.assertEqual(response.status, 400) self.assertIn("email_too_long", response.text) def test_payment_payload_rejects_overlong_description(self): model, response = subscription_webapp._validate_model_payload( subscription_webapp.WebAppPaymentCreatePayload, { "method": "platega", "months": 3, "description": "x" * 4097, }, ) self.assertIsNone(model) self.assertEqual(response.status, 400) self.assertIn("description_too_long", response.text) def test_telegram_oauth_nonce_round_trips(self): settings = SimpleNamespace( WEBAPP_SESSION_SECRET="session-secret", WEBAPP_SESSION_TTL_SECONDS=3600, ) nonce = create_telegram_oauth_nonce(settings, ttl_seconds=60) self.assertTrue(verify_telegram_oauth_nonce(settings, nonce)) self.assertFalse(verify_telegram_oauth_nonce(settings, nonce + "tampered")) async def test_telegram_oauth_start_uses_short_public_state(self): settings = SimpleNamespace( WEBAPP_ENABLED=True, BOT_TOKEN="123456789:secret", TELEGRAM_OAUTH_CLIENT_ID=None, TELEGRAM_OAUTH_CLIENT_SECRET="client-secret", TELEGRAM_OAUTH_REQUEST_ACCESS="write", WEBAPP_LOGIN_TOKEN_TTL_SECONDS=600, WEBAPP_SESSION_SECRET="session-secret", WEBAPP_SESSION_TTL_SECONDS=3600, SUBSCRIPTION_MINI_APP_URL="https://app.example.com/home", ) request = SimpleNamespace( app={"settings": settings}, query={"purpose": "login", "referral_code": "x" * 128}, headers={}, cookies={}, ) with self.assertRaises(web.HTTPFound) as raised: await subscription_webapp.telegram_oauth_start_route(request) redirect = raised.exception query = parse_qs(urlsplit(redirect.headers["Location"]).query) state = query["state"][0] self.assertLessEqual(len(state), 64) cookie_name = subscription_webapp.WEBAPP_TELEGRAM_OAUTH_STATE_COOKIE_NAME signed_state = redirect.cookies[cookie_name].value callback_request = SimpleNamespace( app={"settings": settings}, cookies={cookie_name: signed_state}, ) payload = subscription_webapp._read_telegram_oauth_state_payload(callback_request, state) self.assertIsNotNone(payload) self.assertEqual(payload["referral_code"], "x" * 128) self.assertEqual(len(payload["code_verifier"]), 43) self.assertIsNone( subscription_webapp._read_telegram_oauth_state_payload(callback_request, state + "x") ) def test_telegram_oauth_client_id_defaults_to_bot_id(self): settings = SimpleNamespace( BOT_TOKEN="123456789:secret", TELEGRAM_OAUTH_CLIENT_ID=None, TELEGRAM_OAUTH_REQUEST_ACCESS="write, phone, unknown, write", ) self.assertEqual(subscription_webapp._resolve_telegram_oauth_client_id(settings), 123456789) self.assertEqual( subscription_webapp._resolve_telegram_oauth_request_access(settings), ["write", "phone"], ) def test_public_webapp_base_url_ignores_forwarded_headers_from_untrusted_remote(self): settings = SimpleNamespace( SUBSCRIPTION_MINI_APP_URL="", trusted_proxies=["127.0.0.1"], ) request = SimpleNamespace( remote="203.0.113.10", scheme="http", host="internal.local", headers={ "Host": "internal.local", "X-Forwarded-Proto": "https", "X-Forwarded-Host": "evil.example.com", }, ) self.assertEqual( subscription_webapp._public_webapp_base_url(settings, request), "http://internal.local", ) def test_public_webapp_base_url_accepts_forwarded_headers_from_trusted_proxy(self): settings = SimpleNamespace( SUBSCRIPTION_MINI_APP_URL="", trusted_proxies=["127.0.0.1"], ) request = SimpleNamespace( remote="127.0.0.1", scheme="http", host="internal.local", headers={ "Host": "internal.local", "X-Forwarded-Proto": "https", "X-Forwarded-Host": "app.example.com", }, ) self.assertEqual( subscription_webapp._public_webapp_base_url(settings, request), "https://app.example.com", ) async def test_security_headers_csp_excludes_unsafe_eval_and_plain_http_images(self): request = {"settings": SimpleNamespace()} handler = AsyncMock(return_value=web.Response(text="ok")) response = await subscription_webapp._security_headers_middleware(request, handler) csp = response.headers["Content-Security-Policy"] self.assertNotIn("'unsafe-eval'", csp) self.assertIn("img-src 'self' data: https:;", csp) self.assertNotIn("img-src 'self' data: https: http:;", csp) class AdminSettingsSecurityTests(unittest.IsolatedAsyncioTestCase): async def test_admin_settings_masks_secret_values_and_exposes_has_value(self): class AsyncSessionFactory: def __call__(self): return self async def __aenter__(self): return object() async def __aexit__(self, exc_type, exc, tb): return False settings = Settings( _env_file=None, BOT_TOKEN="token", POSTGRES_USER="app_user", POSTGRES_PASSWORD="app_password", YOOKASSA_SECRET_KEY="super-secret", SHOP_NAME="Visible shop", ) request = SimpleNamespace( app={"settings": settings, "async_session_factory": AsyncSessionFactory()}, headers={}, cookies={}, admin_telegram_id=1, ) request.get = lambda key, default=None: getattr(request, key, default) with ( patch.object(admin_api, "_require_admin_user_id", return_value=1), patch.object( admin_api.app_settings_dal, "get_overrides_with_meta", AsyncMock(return_value=[]), ), ): response = await admin_api.admin_settings_get_route(request) payload = json.loads(response.text) secret_field = next( field for section in payload["sections"] for field in section["fields"] if field["key"] == "YOOKASSA_SECRET_KEY" ) self.assertEqual(secret_field["value"], "") self.assertTrue(secret_field["has_value"]) self.assertNotIn("super-secret", response.text) class DatabaseLoggingSecurityTests(unittest.TestCase): def test_database_url_redaction_hides_password(self): raw_url = "postgresql+asyncpg://user:raw-password@db.example.com:5432/app" redacted = redacted_database_url(raw_url) self.assertIn("user:***@", redacted) self.assertNotIn("raw-password", redacted) def asyncio_run(coro): import asyncio return asyncio.run(coro)