important fix & projscale logger & auth tonconnect

app/api/routes/auth.py

@@ -2,19 +2,25 @@ from datetime import datetime
 
 from aiogram.utils.web_app import safe_parse_webapp_init_data
 from sanic import response
+from sqlalchemy import select, and_
+from tonsdk.utils import Address
 
 from app.core._config import TELEGRAM_API_KEY
+from app.core.logger import make_log
+from app.core.models import KnownKey, WalletConnection
 from app.core.models.user import User
+from pytonconnect.parsers import WalletInfo, Account, TonProof
 
 
 async def s_api_v1_auth_twa(request):
-    if not request.json:
+    auth_data = {}
-        return response.json({"error": "No data provided"}, status=400)
+    for req_key in ['twa_data', 'ton_proof', 'ref_id']:
+        try:
+            auth_data[req_key] = request.json[req_key]
+        except:
+            auth_data[req_key] = None
 
-    if not request.json.get('twa_data'):
+    twa_data = auth_data['twa_data']
-        return response.json({"error": "No TWA data provided"}, status=400)
-
-    twa_data = request.json['twa_data']
     twa_data = safe_parse_webapp_init_data(token=TELEGRAM_API_KEY, init_data=twa_data)
     assert twa_data
 
@@ -39,11 +45,62 @@ async def s_api_v1_auth_twa(request):
         assert known_user, "User not created"
 
     new_user_key = await known_user.create_api_token_v1(request.ctx.db_session, "USER_API_V1")
+    if auth_data['ton_proof']:
+        try:
+            wallet_info = WalletInfo()
+            auth_data['ton_proof']['account']['network'] = auth_data['ton_proof']['account']['chain']
+            wallet_info.account = Account.from_dict(auth_data['ton_proof']['account'])
+            wallet_info.ton_proof = TonProof.from_dict({'proof': auth_data['ton_proof']['ton_proof']})
+            connection_payload = auth_data['ton_proof']['ton_proof']['payload']
+            known_payload = (await request.ctx.db_session.execute(select(KnownKey).where(KnownKey.seed == connection_payload))).scalars().first()
+            assert known_payload, "Unknown payload"
+            assert known_payload.meta['I_user_id'] == known_user.id, "Invalid user_id"
+            assert wallet_info.check_proof(connection_payload), "Invalid proof"
 
-    connected_wallet_data = known_user.wallet_connection(request.ctx.db_session)
+            for known_connection in (await request.ctx.db_session.execute(select(WalletConnection).where(
+                    and_(
+                        WalletConnection.user_id == known_user.id,
+                        WalletConnection.network == 'ton'
+                    )
+            ))).scalars().all():
+                known_connection.invalidated = True
+
+            for other_connection in (await request.ctx.db_session.execute(select(WalletConnection).where(
+                    WalletConnection.wallet_address == Address(wallet_info.account.address).to_string(1, 1, 1)
+            ))).scalars().all():
+                other_connection.invalidated = True
+
+            new_connection = WalletConnection(
+                user_id=known_user.id,
+                network='ton',
+                wallet_key='web2-client==1',
+                connection_id=connection_payload,
+                wallet_address=Address(wallet_info.account.address).to_string(1, 1, 1),
+                keys={
+                    'ton_proof': auth_data['ton_proof']
+                },
+                meta={},
+                created=datetime.now(),
+                updated=datetime.now(),
+                invalidated=False,
+                without_pk=False
+            )
+            request.ctx.db_session.add(new_connection)
+            await request.ctx.db_session.commit()
+        except BaseException as e:
+            make_log("auth", f"Invalid ton_proof: {e}", level="warning")
+            return response.json({"error": "Invalid ton_proof"}, status=400)
+
+    ton_connection = (await request.ctx.db_session.execute(select(WalletConnection).where(
+        and_(
+            WalletConnection.user_id == known_user.id,
+            WalletConnection.network == 'ton',
+            WalletConnection.invalidated == False
+        )
+    ))).scalars().first()
 
     return response.json({
         'user': known_user.json_format(),
-        'connected_wallet': connected_wallet_data.json_format() if connected_wallet_data else None,
+        'connected_wallet': ton_connection.json_format() if ton_connection else None,
         'auth_v1_token': new_user_key['auth_v1_token']
     })

app/core/logger.py

@@ -1,7 +1,6 @@
+import os
+from app.core.projscale_logger import logger
 import logging
-import sys, os
-
-from app.core._config import LOG_LEVEL, LOG_FILEPATH
 
 LOG_LEVELS = {
     'DEBUG': logging.DEBUG,
@@ -9,26 +8,7 @@ LOG_LEVELS = {
     'WARNING': logging.WARNING,
     'ERROR': logging.ERROR
 }
-LOG_LEVEL = LOG_LEVELS[LOG_LEVEL]
+IGNORED_ISSUERS = os.getenv('IGNORED_ISSUERS', '').split(',')
-
-logger = logging.getLogger(__name__)
-logger.setLevel(logging.DEBUG)
-
-handler2 = logging.StreamHandler(sys.stdout)
-handler2.setLevel(LOG_LEVEL)
-handler2.setFormatter(
-    logging.Formatter('%(asctime)s | %(levelname)s | %(message)s')
-)
-logger.addHandler(handler2)
-
-handler3 = logging.FileHandler(LOG_FILEPATH)
-handler3.setLevel(logging.DEBUG)
-handler3.setFormatter(
-    logging.Formatter('%(asctime)s | %(levelname)s | %(message)s')
-)
-logger.addHandler(handler3)
-
-IGNORED_ISSUERS = [v for v in os.getenv('IGNORED_ISSUERS', '').split(',') if v]
 
 
 def make_log(issuer, message, *args, level='INFO', **kwargs):

app/core/projscale_logger.py

@@ -0,0 +1,81 @@
+from datetime import datetime
+import logging
+import time
+import httpx
+import threading
+import os
+
+PROJSCALE_APP_NAME = os.getenv('APP_PROJSCALE_NAME', 'my-uploader')
+LOGS_DIRECTORY = os.getenv('APP_LOGS_DIRECTORY', 'logs')
+os.makedirs(LOGS_DIRECTORY, exist_ok=True)
+
+FORMAT_STRING = '%(asctime)s - %(levelname)s – %(pathname)s – %(funcName)s – %(lineno)d - %(message)s'
+
+
+def push_logs_to_loki(log_entries: list[int, str], attempt: int = 0):
+    try:
+        response = httpx.post(
+            "https://loki-api.projscale.dev/loki/api/v1/push",
+            json={
+                'streams': [
+                    {
+                        'stream': {
+                            'label': 'externalLogs',
+                            'appName': PROJSCALE_APP_NAME
+                        },
+                        'values': log_entries
+                    }
+                ]
+            },
+            headers={
+                'Content-Type': 'application/json'
+            }
+        )
+        assert response.status_code == 204, f"Invalid HTTP status code"
+    except BaseException as e:
+        if attempt < 3:
+            time.sleep(3)
+            push_logs_to_loki(log_entries, attempt + 1)
+        else:
+            print(f"Failed to push logs to Loki: {e}")
+
+
+class ProjscaleLoggingHandler(logging.Handler):
+    def __init__(self):
+        super().__init__()
+        self.setFormatter(logging.Formatter(FORMAT_STRING))
+        self.logs_cache = []
+        self.logs_pushed = 0
+
+    def emit(self, record):
+        log_entry = self.format(record)
+        self.logs_cache.append([str(int(time.time() * 1e9)), log_entry])
+        if ((time.time() - self.logs_pushed) > 5) or (len(self.logs_cache) > 5_000):
+            threading.Thread(target=push_logs_to_loki, args=(self.logs_cache,)).start()
+            self.logs_cache = []
+            self.logs_pushed = time.time()
+
+
+logger = logging.getLogger('uploaderMY')
+
+projscale_handler = ProjscaleLoggingHandler()
+projscale_handler.setLevel(logging.DEBUG)
+logger.addHandler(projscale_handler)
+
+log_filepath = f"{LOGS_DIRECTORY}/{datetime.now().strftime('%Y-%m-%d_%H')}.log"
+file_handler = logging.FileHandler(log_filepath)
+file_handler.setLevel(logging.DEBUG)
+file_handler.setFormatter(logging.Formatter(FORMAT_STRING))
+logger.addHandler(file_handler)
+
+if os.getenv('APP_ENABLE_STDOUT_LOGS', '0') == '1':
+    stdout_handler = logging.StreamHandler()
+    stdout_handler.setLevel(logging.DEBUG)
+    stdout_handler.setFormatter(logging.Formatter(FORMAT_STRING))
+    logger.addHandler(stdout_handler)
+
+
+logging.getLogger("httpx").setLevel(logging.WARNING)
+logging.getLogger("urllib3").setLevel(logging.WARNING)
+logging.getLogger("sanic").setLevel(logging.WARNING)
+logger.setLevel(logging.DEBUG)

app/core/storage.py

@@ -7,8 +7,9 @@ from sqlalchemy.sql import text
 
 from app.core._config import MYSQL_URI, MYSQL_DATABASE
 from app.core.logger import make_log
+from sqlalchemy.pool import NullPool
 
-engine = create_engine(MYSQL_URI)  #, echo=True)
+engine = create_engine(MYSQL_URI, poolclass=NullPool)  #, echo=True)
 Session = sessionmaker(bind=engine)
 
 
@@ -25,7 +26,7 @@ while not database_initialized:
         make_log("SQL", 'MariaDB is not ready yet: ' + str(e), level='debug')
         time.sleep(1)
 
-engine = create_engine(f"{MYSQL_URI}/{MYSQL_DATABASE}")
+engine = create_engine(f"{MYSQL_URI}/{MYSQL_DATABASE}", poolclass=NullPool)
 Session = sessionmaker(bind=engine)