332 lines
11 KiB
Python
332 lines
11 KiB
Python
import os
|
|
import re
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
from sqladmin import Admin, ModelView
|
|
from starlette.datastructures import UploadFile
|
|
from starlette.requests import Request
|
|
from wtforms import BooleanField, FileField, Form, IntegerField, SelectField, StringField
|
|
from wtforms.validators import DataRequired
|
|
|
|
from sqlalchemy import text
|
|
|
|
from app.auth import AdminAuthBackend, get_session_secret
|
|
from app.models import ICON_DIR, SessionLocal, Dapp, Token, Version, Wallet
|
|
|
|
BASE_DIR = Path(__file__).resolve().parent.parent
|
|
TEMPLATES_DIR = str(BASE_DIR / "app" / "templates")
|
|
|
|
MAX_ICON_BYTES = 512 * 1024
|
|
MAX_DAPP_IMAGE_BYTES = 2 * 1024 * 1024
|
|
PNG_MAGIC = b"\x89PNG\r\n\x1a\n"
|
|
PUBLIC_BASE_URL = os.environ.get("IDRS_PUBLIC_URL", "https://idrs.databisnis.id").rstrip("/")
|
|
|
|
DAPP_IMAGE_MAGIC = [
|
|
(b"\x89PNG\r\n\x1a\n", ".png"),
|
|
(b"\xff\xd8\xff", ".jpg"),
|
|
(b"GIF87a", ".gif"),
|
|
(b"GIF89a", ".gif"),
|
|
(b"RIFF", ".webp"),
|
|
(b"\x00\x00\x01\x00", ".ico"),
|
|
(b"BM", ".bmp"),
|
|
]
|
|
|
|
|
|
def _is_authenticated(request: Request) -> bool:
|
|
return bool(request.session.get("auth"))
|
|
|
|
|
|
def _sanitize_png_name(name: str) -> str:
|
|
base = Path(name).name
|
|
if base in ("", ".", ".."):
|
|
return "icon.png"
|
|
if not base.lower().endswith(".png"):
|
|
base = f"{Path(base).stem or 'icon'}.png"
|
|
return base
|
|
|
|
|
|
async def _extract_icon(raw) -> bytes:
|
|
if isinstance(raw, UploadFile):
|
|
return await raw.read(MAX_ICON_BYTES + 1)
|
|
stream = raw.stream if hasattr(raw, "stream") else raw
|
|
data = stream.read(MAX_ICON_BYTES + 1)
|
|
return data if isinstance(data, bytes) else bytes(data)
|
|
|
|
|
|
def _next_token_id() -> int:
|
|
with SessionLocal() as session:
|
|
return int(session.execute(text("SELECT COALESCE(MAX(idfToken), 0) + 1 FROM tokens")).scalar())
|
|
|
|
|
|
def _now_iso() -> str:
|
|
return datetime.now(timezone.utc).isoformat(timespec="seconds")
|
|
|
|
|
|
def _sanitize_dapp_image_name(name: str) -> str:
|
|
base = re.sub(r"[^A-Za-z0-9._-]+", "_", Path(name).name).strip("._") or "dapp"
|
|
return base[:80]
|
|
|
|
|
|
async def _extract_dapp_image(raw) -> bytes:
|
|
if isinstance(raw, UploadFile):
|
|
return await raw.read(MAX_DAPP_IMAGE_BYTES + 1)
|
|
stream = raw.stream if hasattr(raw, "stream") else raw
|
|
data = stream.read(MAX_DAPP_IMAGE_BYTES + 1)
|
|
return data if isinstance(data, bytes) else bytes(data)
|
|
|
|
|
|
def _sniff_image_extension(content: bytes) -> str | None:
|
|
for magic, ext in DAPP_IMAGE_MAGIC:
|
|
if content.startswith(magic):
|
|
return ext
|
|
return None
|
|
|
|
|
|
class NullableIntegerField(IntegerField):
|
|
def process_formdata(self, valuelist):
|
|
if valuelist and valuelist[0].strip() == "":
|
|
self.data = None
|
|
return
|
|
super().process_formdata(valuelist)
|
|
|
|
|
|
class TokenForm(Form):
|
|
position = NullableIntegerField("Position", default=_next_token_id)
|
|
name = StringField("Name", validators=[DataRequired()])
|
|
symbol = StringField("Symbol", validators=[DataRequired()])
|
|
contractAddr = StringField("Contract", validators=[DataRequired()])
|
|
decimals = IntegerField("Decimals", validators=[DataRequired()])
|
|
iconUrl = StringField("Icon URL")
|
|
icon = FileField("Icon PNG")
|
|
is_live = BooleanField("Live")
|
|
|
|
|
|
class TokenAdmin(ModelView, model=Token):
|
|
name = "Tokens"
|
|
icon = "fa-solid fa-coins"
|
|
form = TokenForm
|
|
column_list = [
|
|
"position",
|
|
"name",
|
|
"symbol",
|
|
"typeBlockchain",
|
|
"contractAddr",
|
|
"decimals",
|
|
"is_live",
|
|
"iconUrl",
|
|
]
|
|
column_labels = {
|
|
"idfToken": "ID",
|
|
"position": "Position",
|
|
"typeBlockchain": "Blockchain",
|
|
"contractAddr": "Contract",
|
|
"decimals": "Decimals",
|
|
"is_live": "Live",
|
|
}
|
|
column_details_list = [
|
|
"idfToken",
|
|
"position",
|
|
"typeBlockchain",
|
|
"name",
|
|
"symbol",
|
|
"contractAddr",
|
|
"decimals",
|
|
"iconUrl",
|
|
"filename",
|
|
"is_live",
|
|
]
|
|
column_searchable_list = ["name", "symbol", "contractAddr"]
|
|
column_default_sort = [("position", True)]
|
|
page_size = 25
|
|
|
|
def is_accessible(self, request: Request) -> bool:
|
|
return _is_authenticated(request)
|
|
|
|
async def on_model_change(self, data, model, is_created, request):
|
|
data.pop("idfToken", None)
|
|
data["typeBlockchain"] = "Vexanium"
|
|
if data.get("position") in (None, ""):
|
|
data["position"] = _next_token_id()
|
|
raw = data.get("icon")
|
|
data.pop("icon", None)
|
|
if not raw:
|
|
return
|
|
content = await _extract_icon(raw)
|
|
if not content:
|
|
return
|
|
if len(content) > MAX_ICON_BYTES:
|
|
raise Exception(f"Icon must be PNG ≤ {MAX_ICON_BYTES // 1024}KB")
|
|
if not content.startswith(PNG_MAGIC):
|
|
raise Exception("Icon must be a real PNG file")
|
|
filename = _sanitize_png_name(getattr(raw, "filename", "") or "icon.png")
|
|
token_dir = ICON_DIR / "token"
|
|
token_dir.mkdir(parents=True, exist_ok=True)
|
|
(token_dir / filename).write_bytes(content)
|
|
data["filename"] = filename
|
|
data["iconUrl"] = f"{PUBLIC_BASE_URL}/img/token/{filename}"
|
|
|
|
|
|
class DappForm(Form):
|
|
name = StringField("Name", validators=[DataRequired()])
|
|
description = StringField("Description")
|
|
link = StringField("Link")
|
|
image = StringField("Image URL")
|
|
imageFile = FileField("Image file")
|
|
is_live = BooleanField("Live")
|
|
counter = IntegerField("Counter", default=0)
|
|
category_id = NullableIntegerField("Category ID")
|
|
is_available_indonesia = SelectField(
|
|
"Available in Indonesia",
|
|
choices=[("", "—"), ("true", "Yes"), ("false", "No")],
|
|
default="",
|
|
)
|
|
is_new_version = BooleanField("New Version")
|
|
|
|
|
|
class DappAdmin(ModelView, model=Dapp):
|
|
name = "Dapps"
|
|
icon = "fa-solid fa-globe"
|
|
form = DappForm
|
|
column_list = [
|
|
"name",
|
|
"link",
|
|
"image",
|
|
"is_live",
|
|
"counter",
|
|
"category_id",
|
|
"is_available_indonesia",
|
|
"is_new_version",
|
|
]
|
|
column_labels = {
|
|
"is_live": "Live",
|
|
"category_id": "Category",
|
|
"is_available_indonesia": "Avail. ID",
|
|
}
|
|
column_details_list = [
|
|
"id",
|
|
"name",
|
|
"description",
|
|
"link",
|
|
"image",
|
|
"is_live",
|
|
"counter",
|
|
"category_id",
|
|
"is_available_indonesia",
|
|
"is_new_version",
|
|
"created_at",
|
|
"updated_at",
|
|
]
|
|
column_searchable_list = ["name", "description", "link"]
|
|
column_default_sort = [("id", True)]
|
|
page_size = 25
|
|
|
|
def is_accessible(self, request: Request) -> bool:
|
|
return _is_authenticated(request)
|
|
|
|
async def on_model_change(self, data, model, is_created, request):
|
|
data.pop("id", None)
|
|
data.pop("is_live_override", None)
|
|
submitted_live = bool(data.get("is_live"))
|
|
if is_created:
|
|
data["is_live_override"] = not submitted_live
|
|
elif submitted_live != bool(model.is_live):
|
|
data["is_live_override"] = not submitted_live
|
|
if data.get("is_available_indonesia") == "":
|
|
data["is_available_indonesia"] = None
|
|
elif data.get("is_available_indonesia") in ("true", "True", "1", 1, "y", "on"):
|
|
data["is_available_indonesia"] = True
|
|
elif data.get("is_available_indonesia") in ("false", "False", "0", 0, "n", "off"):
|
|
data["is_available_indonesia"] = False
|
|
raw = data.get("imageFile")
|
|
data.pop("imageFile", None)
|
|
if raw:
|
|
content = await _extract_dapp_image(raw)
|
|
if not content:
|
|
return
|
|
if len(content) > MAX_DAPP_IMAGE_BYTES:
|
|
raise Exception(f"Image must be ≤ {MAX_DAPP_IMAGE_BYTES // 1024}KB")
|
|
ext = _sniff_image_extension(content)
|
|
if not ext:
|
|
raise Exception("Image must be a real image (PNG/JPEG/GIF/WebP/ICO/BMP)")
|
|
filename = f"{Path(_sanitize_dapp_image_name(getattr(raw, 'filename', '') or 'dapp')).stem}{ext}"
|
|
dapp_dir = ICON_DIR / "dapps"
|
|
dapp_dir.mkdir(parents=True, exist_ok=True)
|
|
(dapp_dir / filename).write_bytes(content)
|
|
data["image"] = f"{PUBLIC_BASE_URL}/img/dapps/{filename}"
|
|
if is_created:
|
|
data["created_at"] = data.get("created_at") or _now_iso()
|
|
data["updated_at"] = _now_iso()
|
|
|
|
|
|
class VersionAdmin(ModelView, model=Version):
|
|
name = "Android Version"
|
|
icon = "fa-solid fa-mobile-screen"
|
|
can_create = False
|
|
can_delete = False
|
|
column_list = ["idVersion", "appName", "acceptableVersion", "latestVersion"]
|
|
column_labels = {
|
|
"idVersion": "ID Version",
|
|
"appName": "App",
|
|
"acceptableVersion": "Acceptable",
|
|
"latestVersion": "Latest",
|
|
}
|
|
column_details_list = ["idVersion", "appName", "acceptableVersion", "latestVersion"]
|
|
form_columns = ["idVersion", "appName", "acceptableVersion", "latestVersion"]
|
|
|
|
def is_accessible(self, request: Request) -> bool:
|
|
return _is_authenticated(request)
|
|
|
|
async def on_model_change(self, data, model, is_created, request):
|
|
acceptable = int(data.get("acceptableVersion", model.acceptableVersion))
|
|
latest = int(data.get("latestVersion", model.latestVersion))
|
|
if acceptable > latest:
|
|
raise Exception("acceptableVersion must be <= latestVersion")
|
|
|
|
|
|
class WalletForm(Form):
|
|
account = StringField("Account", validators=[DataRequired()])
|
|
|
|
|
|
class WalletAdmin(ModelView, model=Wallet):
|
|
name = "Whitelisted Wallets"
|
|
icon = "fa-solid fa-wallet"
|
|
form = WalletForm
|
|
column_list = ["account", "created_at"]
|
|
column_details_list = ["id", "account", "created_at"]
|
|
column_searchable_list = ["account"]
|
|
column_default_sort = [("id", True)]
|
|
page_size = 25
|
|
|
|
def is_accessible(self, request: Request) -> bool:
|
|
return _is_authenticated(request)
|
|
|
|
async def on_model_change(self, data, model, is_created, request):
|
|
data.pop("id", None)
|
|
account = (data.get("account") or "").strip()
|
|
if not account:
|
|
raise Exception("Account is required")
|
|
with SessionLocal() as session:
|
|
exists = session.query(Wallet).filter(Wallet.account == account).first()
|
|
if exists and (is_created or exists.id != model.id):
|
|
raise Exception(f"'{account}' is already whitelisted")
|
|
data["account"] = account
|
|
if is_created:
|
|
data["created_at"] = _now_iso()
|
|
|
|
|
|
def create_admin(app):
|
|
auth_backend = AdminAuthBackend(get_session_secret())
|
|
admin = Admin(
|
|
app,
|
|
session_maker=SessionLocal,
|
|
title="IDRS Admin",
|
|
base_url="/admin",
|
|
templates_dir=TEMPLATES_DIR,
|
|
authentication_backend=auth_backend,
|
|
)
|
|
admin.add_view(TokenAdmin)
|
|
admin.add_view(VersionAdmin)
|
|
admin.add_view(DappAdmin)
|
|
admin.add_view(WalletAdmin)
|
|
return admin |