import os import json import hashlib import datetime import time import uuid import pytz # 处理多伦多时区 from fastapi import FastAPI, Request, Response from fastapi.responses import PlainTextResponse, JSONResponse from fastapi.middleware.cors import CORSMiddleware import gspread from google.oauth2.service_account import Credentials import uvicorn from apscheduler.schedulers.background import BackgroundScheduler from apscheduler.triggers.cron import CronTrigger app = FastAPI() app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"], ) # ===================== 配置 ===================== SHEET_ID = os.getenv('SHEET_ID') TOKEN_EXPIRE_MIN = 15 TORONTO_TZ = pytz.timezone('America/Toronto') T_WINDOW = 10 # 允许的时间误差：10秒 #Transfer_Cooldown_Card = {} #Transfer_Cooldown = {} TRANSFER_LIMIT = 1500 # 普通转账单次限额 CARD_PAY_LIMIT = 200 # 刷卡转账单次限额 # --- 防 DDoS 配置 --- visit_history = {} BLACKLIST = os.getenv("BLACKLIST_IPS", "").split(",") def get_gc_client(): creds_json = os.getenv("G_CREDS") info = json.loads(creds_json) creds = Credentials.from_service_account_info(info, scopes=[ "https://www.googleapis.com/auth/spreadsheets", "https://www.googleapis.com/auth/drive" ]) return gspread.authorize(creds) gc = get_gc_client() spreadsheet = gc.open_by_key(SHEET_ID) # ===================== 核心防御中间件 ===================== def is_rate_limited(ip: str): now = time.time() if ip in [x.strip() for x in BLACKLIST if x.strip()]: return True if ip not in visit_history: visit_history[ip] = [] visit_history[ip] = [t for t in visit_history[ip] if now - t < 10] if len(visit_history[ip]) >= 30: return True visit_history[ip].append(now) return False @app.middleware("http") async def ddos_protection_middleware(request: Request, call_next): forwarded = request.headers.get("X-Forwarded-For") ip = forwarded.split(",")[0].strip() if forwarded else request.client.host if is_rate_limited(ip): return PlainTextResponse("error: too many requests", status_code=429) response = await call_next(request) return response @app.middleware("http") async def add_security_headers(request, call_next): response = await call_next(request) # 1. 防嗅探：告诉浏览器必须听服务器的 Content-Type response.headers["X-Content-Type-Options"] = "nosniff" # 2. 防点击劫持：禁止或限制网页被嵌入 iFrame response.headers["X-Frame-Options"] = "DENY" # 3. 开启 HSTS：强制走 HTTPS (Hugging Face 默认支持) response.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains" # 4. 内容安全策略 (CSP)：最强防线 # 这里先设置一个较宽松的，允许加载自家的脚本和样式 response.headers["Content-Security-Policy"] = "default-src 'self'; script-src 'self' 'unsafe-inline'; style-src 'self' 'unsafe-inline';" # 5. 权限控制：禁用摄像头、麦克风等（如果你用不到） response.headers["Permissions-Policy"] = "camera=(), microphone=(), geolocation=()" return response # ===================== 工具函数 ===================== def get_now_toronto(): return datetime.datetime.now(TORONTO_TZ).strftime('%Y-%m-%d %H:%M:%S') def write_system_log(user: str, action: str, status: str, request: Request): try: sh = spreadsheet.worksheet('SystemLogs') now = get_now_toronto() forwarded = request.headers.get("X-Forwarded-For") ip = forwarded.split(",")[0].strip() if forwarded else request.client.host sh.append_row([now, user, action, status, ip]) except: pass def create_session_in_sheet(username: str): sh = spreadsheet.worksheet('Sessions') token = str(uuid.uuid4()) # 修改这里：使用带时区的当前时间 now_utc = datetime.datetime.now(datetime.timezone.utc) expire_at = (now_utc + datetime.timedelta(minutes=TOKEN_EXPIRE_MIN)).isoformat() rows = sh.get_all_values() updated = False for i, row in enumerate(rows[1:], start=2): if row[1] == username: sh.update(values=[[token, username, expire_at]], range_name=f"A{i}:C{i}") updated = True break if not updated: sh.append_row([token, username, expire_at]) return token def read_account(username: str): try: sh = spreadsheet.worksheet('Accounts') rows = sh.get_all_values() for i, row in enumerate(rows[1:], start=2): if row[0].strip() == username.strip(): # 补全列，防止读取老数据时越界 row = row + ["0"] * (10 - len(row)) return { "rowIndex": i, "username": row[0], "passwordHex": row[1], "money": float(row[2] or 0), "logs": json.loads(row[3]) if row[3] else [], "admin": str(row[4]).upper() == "TRUE", "email": row[5] if len(row) > 5 else "", "cardHash": row[6] if len(row) > 6 else "", "payPinHash": row[7] if len(row) > 7 else "", "dailyTransTotal": float(row[8] or 0), # I列 "dailyCardTotal": float(row[9] or 0) # J列 } except: return None return None def find_account_by_card_hash(card_hash: str): try: sh = spreadsheet.worksheet('Accounts') rows = sh.get_all_values() for i, row in enumerate(rows[1:], start=2): if len(row) >= 7 and row[6].strip() == card_hash.strip(): return read_account(row[0]) except: return None return None # ===================== 路由接口 ===================== @app.get("/") @app.head("/") async def heartbeat(): return PlainTextResponse("API Active: 10s Security Window Enabled") @app.get("/ping") async def ping(): return PlainTextResponse("ping ok") # 返回服务器时间，方便Scratch同步 #@app.get("/register") async def register(username: str, passwordHex: str, pin: str, request: Request): if read_account(username): write_system_log(username, "REGISTER", "FAIL: EXISTS", request) return PlainTextResponse("username wrong") generated_card_hash = str(uuid.uuid4()) sh = spreadsheet.worksheet('Accounts') # 新账号自动补全 I 列和 J 列的初始值 0 sh.append_row([username, passwordHex, 0, "[]", "FALSE", "", generated_card_hash, pin, 0, 0]) write_system_log(username, "REGISTER", "SUCCESS", request) return PlainTextResponse(f"register ok|{generated_card_hash}") @app.get("/getinfo_by_card") async def getinfo_by_card(card_hash: str, request: Request): acc = find_account_by_card_hash(card_hash) if not acc: return PlainTextResponse("error: invalid card") return JSONResponse({"ok": True, "username": acc["username"], "money": acc["money"], "admin": acc["admin"], "logs": acc["logs"]}) @app.get("/transfer_by_card") async def transfer_by_card(card_hash: str, pin: str, t: str, to: str, money: str, request: Request): try: if abs(int(time.time()) - int(t)) > T_WINDOW: return PlainTextResponse("error: expired") except: return PlainTextResponse("error: invalid time") acc_f = find_account_by_card_hash(card_hash) if not acc_f: return PlainTextResponse("error: invalid card") # 动态加盐校验：sha256(t + 数据库存的PIN_HEX) expected_sign = hashlib.sha256(f"{t}{acc_f['payPinHash']}".encode()).hexdigest() if pin.strip() != expected_sign: write_system_log(acc_f["username"], "CARD_PAY", "FAIL: WRONG DYNAMIC PIN", request) return PlainTextResponse("error: wrong pin") acc_t = read_account(to) if not acc_t: return PlainTextResponse("error: to user wrong") if acc_f["username"] == acc_t["username"]: return PlainTextResponse("error: to user wrong") amount = float(money) if not acc_f["admin"]: if amount <= 0 or acc_f["money"] < amount: return PlainTextResponse("error: money wrong") # 每日刷卡限额校验 (J列) if acc_f["dailyCardTotal"] + amount > CARD_PAY_LIMIT: return PlainTextResponse("error: money wrong") acc_f["money"] = round(acc_f["money"] - amount, 2) acc_t["money"] = round(acc_t["money"] + amount, 2) # 累加每日卡交易金额 acc_f["dailyCardTotal"] = round(acc_f["dailyCardTotal"] + amount, 2) local_time = get_now_toronto() acc_f["logs"].append(f"{local_time} Card Paid Đ{money} to {to}.") acc_t["logs"].append(f"{local_time} Received Đ{money} from Card({acc_f['username']}).") sh_acc = spreadsheet.worksheet('Accounts') for acc in [acc_f, acc_t]: row_data = [ acc['username'], acc['passwordHex'], acc['money'], json.dumps(acc['logs']), str(acc['admin']).upper(), acc['email'], acc['cardHash'], acc['payPinHash'], acc['dailyTransTotal'], acc['dailyCardTotal'] ] # 更新范围扩大到 J 列 sh_acc.update(values=[row_data], range_name=f"A{acc['rowIndex']}:J{acc['rowIndex']}") write_system_log(acc_f["username"], "CARD_PAY", f"SUCCESS: {money} TO {to}", request) return PlainTextResponse("pay ok") @app.get("/login") async def login(username: str, t: str, sign: str, request: Request): acc = read_account(username) if not acc: return PlainTextResponse("wrong") expected = hashlib.sha256(f"{username}{t}{acc['passwordHex']}".encode()).hexdigest() if sign == expected and abs(int(time.time()) - int(t)) < T_WINDOW: token = create_session_in_sheet(username) write_system_log(username, "LOGIN", "SUCCESS", request) return PlainTextResponse(token) write_system_log(username, "LOGIN", "Failed", request) return PlainTextResponse("error: bad sign") @app.get("/getinfo") async def get_info(t: str, sign: str, request: Request): sh_s = spreadsheet.worksheet('Sessions') s_rows = sh_s.get_all_values() # 修改这里：使用带时区的当前时间 now = datetime.datetime.now(datetime.timezone.utc) for row in s_rows[1:]: token, user, expire_str = row[0], row[1], row[2] # 修改这里：解析时确保包含时区信息 expire_dt = datetime.datetime.fromisoformat(expire_str) if expire_dt.tzinfo is None: expire_dt = expire_dt.replace(tzinfo=datetime.timezone.utc) if now > expire_dt: continue if hashlib.sha256(f"{token}{t}".encode()).hexdigest() == sign: acc = read_account(user) return JSONResponse({"ok": True, "money": acc["money"], "admin": acc["admin"], "logs": acc["logs"]}) return PlainTextResponse("error: invalid session") @app.get("/transfer") async def transfer(t: str, sign: str, to: str, money: str, request: Request): sh_s = spreadsheet.worksheet('Sessions') s_rows = sh_s.get_all_values() # 强制获取带 UTC 标记的当前时间 now = datetime.datetime.now(datetime.timezone.utc) for row in s_rows[1:]: token, user, expire_str = row[0], row[1], row[2] # 修复：解析过期时间并强制对齐时区 expire_dt = datetime.datetime.fromisoformat(expire_str) if expire_dt.tzinfo is None: expire_dt = expire_dt.replace(tzinfo=datetime.timezone.utc) if now > expire_dt: continue acc_f = read_account(user) if not acc_f: continue if hashlib.sha256(f"{token}{t}{acc_f['passwordHex']}".encode()).hexdigest() == sign: acc_t = read_account(to) if not acc_t or (acc_t == acc_f and not acc_f["admin"]): return PlainTextResponse("error: to user wrong") amount = float(money) if not acc_f["admin"]: if amount <= 0 or acc_f["money"] < amount: return PlainTextResponse("error: money wrong") # 每日普通转账限额校验 (I列) if acc_f["dailyTransTotal"] + amount > TRANSFER_LIMIT: return PlainTextResponse("error: money wrong") # 执行转账并保留两位小数 acc_f["money"] = round(acc_f["money"] - amount, 2) acc_t["money"] = round(acc_t["money"] + amount, 2) # 累加每日转账金额 acc_f["dailyTransTotal"] = round(acc_f["dailyTransTotal"] + amount, 2) local_time = get_now_toronto() acc_f["logs"].append(f"{local_time} Sent Đ{money} to {to}.") acc_t["logs"].append(f"{local_time} Received Đ{money} from {user}.") sh_acc = spreadsheet.worksheet('Accounts') for acc in [acc_f, acc_t]: row_data = [ acc['username'], acc['passwordHex'], acc['money'], json.dumps(acc['logs']), str(acc['admin']).upper(), acc['email'], acc['cardHash'], acc['payPinHash'], acc['dailyTransTotal'], acc['dailyCardTotal'] ] # 更新范围扩大到 J 列 sh_acc.update(values=[row_data], range_name=f"A{acc['rowIndex']}:J{acc['rowIndex']}") write_system_log(acc_f["username"], "Transfer", f"SUCCESS: {money} TO {to}", request) return PlainTextResponse("transfer ok") return PlainTextResponse("error: security check failed") @app.get("/logintoken") async def login_token(t: str, sign: str, request: Request): sh_s = spreadsheet.worksheet('Sessions') s_rows = sh_s.get_all_values() # 1. 强制获取带 UTC 时区的当前时间 now = datetime.datetime.now(datetime.timezone.utc) # 2. 校验请求时间窗口（防止重放攻击） try: if abs(int(time.time()) - int(t)) > T_WINDOW: return PlainTextResponse("error: expired") except: return PlainTextResponse("error: invalid time") for row in s_rows[1:]: # 补齐列数，防止索引错误 row = row + [""] * (3 - len(row)) token, user, expire_str = row[0], row[1], row[2] # 3. 修复核心报错：解析过期时间并强制对齐 UTC 时区 try: expire_dt = datetime.datetime.fromisoformat(expire_str) if expire_dt.tzinfo is None: expire_dt = expire_dt.replace(tzinfo=datetime.timezone.utc) except: continue # 如果时间格式损坏，跳过此行 # 4. 执行比较 if now > expire_dt: continue # 5. 校验签名：sha256(token + t) if hashlib.sha256(f"{token}{t}".encode()).hexdigest() == sign: return PlainTextResponse(user) return PlainTextResponse("error: valid") @app.get("/logout") async def logout(t: str, sign: str, request: Request): sh_s = spreadsheet.worksheet('Sessions') s_rows = sh_s.get_all_values() for i, row in enumerate(s_rows[1:], start=2): token, user = row[0], row[1] if hashlib.sha256(f"{token}{t}".encode()).hexdigest() == sign: sh_s.delete_rows(i) return PlainTextResponse("logout ok") return PlainTextResponse("error: invalid session") # ===================== 利息与限额调度核心逻辑 ===================== def apply_weekly_interest(): try: print(f"[{get_now_toronto()}] >>> STARTING INTEREST JOB (Positive & Negative) <<<") sh = spreadsheet.worksheet('Accounts') all_data = sh.get_all_values() if len(all_data) <= 1: return rows = all_data[1:] updated_data = [] interest_rate = 1.0375 now_time = get_now_toronto() for row in rows: # 补齐到 10 列，防止越界 row = row + ["0"] * (10 - len(row)) try: current_money = float(row[2] or 0) # 只要余额不为 0（包括负数），就计算利息 if current_money != 0: new_money = round(current_money * interest_rate, 2) interest_diff = round(new_money - current_money, 2) # 鲁棒解析日志 JSON try: logs = json.loads(row[3]) if (row[3] and row[3].startswith('[')) else [] except: logs = [] # 符号处理：正数显示 +，负数本身自带 - prefix = "+" if interest_diff > 0 else "" logs.append(f"{now_time} Weekly Interest: {prefix}Đ{interest_diff}") row[2] = str(new_money) # 转回字符串存入表格 row[3] = json.dumps(logs) updated_data.append(row) except Exception as e: print(f"Row skip error: {e}") updated_data.append(row) if updated_data: # 扩展到 J 列更新 sh.update(values=updated_data, range_name=f'A2:J{len(updated_data) + 1}') print(f"[{get_now_toronto()}] >>> INTEREST JOB SUCCESSFUL <<<") except Exception as e: print(f"CRITICAL ERROR in interest job: {e}") def reset_daily_limits(): """每天 00:00 重置所有用户的 I 和 J 列""" try: print(f"[{get_now_toronto()}] >>> STARTING DAILY LIMIT RESET <<<") sh = spreadsheet.worksheet('Accounts') all_data = sh.get_all_values() if len(all_data) <= 1: return row_count = len(all_data) # 准备两列全为 0 的数据用于覆盖 (从第2行开始) zero_data = [["0", "0"]] * (row_count - 1) # 批量更新 I2:J{最后一行} sh.update(values=zero_data, range_name=f"I2:J{row_count}") print(f"[{get_now_toronto()}] >>> DAILY LIMIT RESET SUCCESSFUL <<<") except Exception as e: print(f"CRITICAL ERROR in reset limits job: {e}") # ===================== 自动化调度器配置 ===================== scheduler = BackgroundScheduler(timezone=TORONTO_TZ) # 每周五利息结算 trigger = CronTrigger(day_of_week='fri', hour=17, minute=30) scheduler.add_job(apply_weekly_interest, trigger) # 每天凌晨 00:00 执行每日限额清零 reset_trigger = CronTrigger(hour=4, minute=0) scheduler.add_job(reset_daily_limits, reset_trigger) scheduler.start() print("Scheduler started: Weekly interest will be applied every Friday at 17:30 EST.") print("Scheduler started: Daily limits will be reset every day at 00:00 EST.") if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=7860)