58 lines
1.6 KiB
Python
58 lines
1.6 KiB
Python
import logging
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.exc import IntegrityError
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.core.config import get_settings
|
|
|
|
logger = logging.getLogger("gsfdc.ip")
|
|
|
|
|
|
def _blocked_by_env(ip: str) -> bool:
|
|
return ip in get_settings().blocked_ips
|
|
|
|
|
|
def is_ip_blocked(ip: str) -> bool:
|
|
if not (get_settings().database_url or "").strip():
|
|
return _blocked_by_env(ip)
|
|
try:
|
|
from app.db import SessionLocal
|
|
from app.models.security import AdminIpBlacklist
|
|
|
|
db = SessionLocal()
|
|
try:
|
|
found = db.scalar(select(AdminIpBlacklist.id).where(AdminIpBlacklist.ip == ip))
|
|
finally:
|
|
db.close()
|
|
return found is not None
|
|
except Exception as exc:
|
|
logger.warning("admin ip blacklist lookup failed: %s", type(exc).__name__)
|
|
return _blocked_by_env(ip)
|
|
|
|
|
|
def seed_ip_blacklist(db: Session) -> None:
|
|
from app.models.security import AdminIpBlacklist
|
|
|
|
wanted = set(get_settings().blocked_ips)
|
|
if not wanted:
|
|
return
|
|
existing = set(db.scalars(select(AdminIpBlacklist.ip)).all())
|
|
pending = [ip for ip in wanted if ip not in existing]
|
|
if not pending:
|
|
return
|
|
for ip in pending:
|
|
db.add(AdminIpBlacklist(ip=ip, remark=""))
|
|
try:
|
|
db.commit()
|
|
except IntegrityError:
|
|
db.rollback()
|
|
for ip in pending:
|
|
if db.scalar(select(AdminIpBlacklist.id).where(AdminIpBlacklist.ip == ip)) is not None:
|
|
continue
|
|
db.add(AdminIpBlacklist(ip=ip, remark=""))
|
|
try:
|
|
db.commit()
|
|
except IntegrityError:
|
|
db.rollback()
|