76 lines
1.8 KiB
Python
76 lines
1.8 KiB
Python
from fastapi import FastAPI, Request, Depends
|
|
from fastapi.responses import HTMLResponse
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy import select
|
|
from pathlib import Path
|
|
|
|
from .db import AsyncSessionLocal, engine, Base
|
|
from .models import Fingerprint
|
|
from .utils import hash_fingerprint
|
|
|
|
app = FastAPI()
|
|
|
|
|
|
@app.get("/", response_class=HTMLResponse)
|
|
async def root():
|
|
static_path = Path(__file__).parent / "static" / "index.html"
|
|
return static_path.read_text()
|
|
|
|
|
|
# Dependency
|
|
async def get_db():
|
|
async with AsyncSessionLocal() as session:
|
|
yield session
|
|
|
|
|
|
@app.on_event("startup")
|
|
async def startup():
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
|
|
|
|
@app.post("/fingerprint")
|
|
async def collect_fingerprint(request: Request, db: AsyncSession = Depends(get_db)):
|
|
body = await request.json()
|
|
|
|
fp_hash = hash_fingerprint(body)
|
|
|
|
ip = request.client.host
|
|
user_agent = request.headers.get("user-agent", "")
|
|
|
|
record = Fingerprint(
|
|
fingerprint_hash=fp_hash,
|
|
raw_data=body,
|
|
ip_address=ip,
|
|
user_agent=user_agent,
|
|
)
|
|
|
|
db.add(record)
|
|
await db.commit()
|
|
|
|
return {
|
|
"status": "stored",
|
|
"fingerprint_hash": fp_hash,
|
|
}
|
|
|
|
|
|
@app.get("/fingerprint/{fp_hash}")
|
|
async def get_fingerprint(fp_hash: str, db: AsyncSession = Depends(get_db)):
|
|
result = await db.execute(
|
|
select(Fingerprint).where(Fingerprint.fingerprint_hash == fp_hash)
|
|
)
|
|
|
|
rows = result.scalars().all()
|
|
|
|
return {
|
|
"count": len(rows),
|
|
"records": [
|
|
{
|
|
"ip": r.ip_address,
|
|
"created_at": r.created_at,
|
|
"user_agent": r.user_agent,
|
|
}
|
|
for r in rows
|
|
],
|
|
}
|