diff --git a/src/core/database.py b/src/core/database.py index dc90389..737fd86 100644 --- a/src/core/database.py +++ b/src/core/database.py @@ -2,7 +2,7 @@ import os from datetime import datetime from typing import Optional -from sqlalchemy import Boolean, Double, String, Text, create_engine +from sqlalchemy import Boolean, Double, String, Text, create_engine, text from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, sessionmaker from core.config import DATA_DIR, DATABASE_URL @@ -20,12 +20,14 @@ class GateDB(Base): id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True) name: Mapped[str] = mapped_column(String, nullable=False) - gate_type: Mapped[str] = mapped_column(String, nullable=False) # 'car' | 'pedestrian' - avconnect_macro_id: Mapped[str] = mapped_column(String, nullable=False) # AVConnect macro ID - status: Mapped[str] = mapped_column(String, default="enabled") # 'enabled' | 'disabled' - group_name: Mapped[Optional[str]] = mapped_column(String, nullable=True) # display group label - lat: Mapped[Optional[float]] = mapped_column(Double, nullable=True) # WGS-84 latitude - lon: Mapped[Optional[float]] = mapped_column(Double, nullable=True) # WGS-84 longitude + gate_type: Mapped[str] = mapped_column(String, nullable=False) # 'car' | 'pedestrian' + api_provider: Mapped[str] = mapped_column(String, nullable=False, default="avconnect") # 'avconnect' | 'shelly' + avconnect_macro_id: Mapped[Optional[str]] = mapped_column(String, nullable=True) # AVConnect macro ID + shelly_device_id: Mapped[Optional[str]] = mapped_column(String, nullable=True) # Shelly Cloud device ID + status: Mapped[str] = mapped_column(String, default="enabled") # 'enabled' | 'disabled' + group_name: Mapped[Optional[str]] = mapped_column(String, nullable=True) # display group label + lat: Mapped[Optional[float]] = mapped_column(Double, nullable=True) # WGS-84 latitude + lon: Mapped[Optional[float]] = mapped_column(Double, nullable=True) # WGS-84 longitude class ApiCredential(Base): @@ -38,6 +40,14 @@ class ApiCredential(Base): mock_avconnect: Mapped[bool] = mapped_column(Boolean, default=False) +class ShellyCredential(Base): + __tablename__ = "shelly_credentials" + + id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True) + server_uri: Mapped[str] = mapped_column(String, nullable=False) # e.g. https://shelly-3.eu.shelly.cloud + auth_key_enc: Mapped[str] = mapped_column(String, nullable=False) # Fernet-encrypted auth key + + class Keypass(Base): __tablename__ = "keypasses" diff --git a/src/core/schemas.py b/src/core/schemas.py index 616c98a..4299614 100644 --- a/src/core/schemas.py +++ b/src/core/schemas.py @@ -81,7 +81,9 @@ class GateResponse(BaseModel): id: int name: str gate_type: str - avconnect_macro_id: str + api_provider: str + avconnect_macro_id: Optional[str] = None + shelly_device_id: Optional[str] = None status: str group_name: Optional[str] = None lat: Optional[float] = None @@ -101,15 +103,17 @@ class GatePublicResponse(BaseModel): class GateCreate(BaseModel): name: str - gate_type: str # 'car' | 'pedestrian' - avconnect_macro_id: str + gate_type: str # 'car' | 'pedestrian' + api_provider: str = "avconnect" # 'avconnect' | 'shelly' + avconnect_macro_id: Optional[str] = None + shelly_device_id: Optional[str] = None status: str = "enabled" group_name: Optional[str] = None lat: Optional[float] = None lon: Optional[float] = None -# ── AVConnect Credentials ───────────────────────────────────────────────────── +# ── API Credentials ────────────────────────────────────────────────────────── class CredentialRead(BaseModel): id: int @@ -121,6 +125,16 @@ class CredentialUpsert(BaseModel): password: str +class ShellyCredentialRead(BaseModel): + id: int + server_uri: str + + +class ShellyCredentialUpsert(BaseModel): + server_uri: str + auth_key: str + + # ── Admin users ─────────────────────────────────────────────────────────────── class AdminUserResponse(BaseModel): diff --git a/src/routers/credentials.py b/src/routers/credentials.py index ebf866a..d29eba6 100644 --- a/src/routers/credentials.py +++ b/src/routers/credentials.py @@ -5,14 +5,17 @@ from pydantic import BaseModel from sqlalchemy.orm import Session from core.auth import encrypt_secret -from core.database import ApiCredential, get_db +from core.database import ApiCredential, ShellyCredential, get_db from core.dependencies import require_admin -from core.schemas import CredentialRead, CredentialUpsert +from core.schemas import CredentialRead, CredentialUpsert, ShellyCredentialRead, ShellyCredentialUpsert from services.avconnect import AVConnectAPI +from services.shelly import ShellyCloudAPI router = APIRouter(prefix="/api/admin/credentials", tags=["admin-credentials"]) +# ── AVConnect credentials ───────────────────────────────────────────────────── + @router.get("", response_model=list[CredentialRead]) async def list_credentials( db: Session = Depends(get_db), _: dict = Depends(require_admin) @@ -37,7 +40,7 @@ async def upsert_credential( if cred: cred.username = req.username cred.password_enc = encrypt_secret(req.password) - cred.session_id = session_id # reuse the session obtained during validation + cred.session_id = session_id else: cred = ApiCredential( username=req.username, @@ -50,7 +53,47 @@ async def upsert_credential( return CredentialRead(id=cred.id, username=cred.username) -# ── Mock AVConnect setting ───────────────────────────────────────────────────── +# ── Shelly Cloud credentials ────────────────────────────────────────────────── + +@router.get("/shelly", response_model=Optional[ShellyCredentialRead]) +async def get_shelly_credential( + db: Session = Depends(get_db), _: dict = Depends(require_admin) +): + cred: Optional[ShellyCredential] = db.query(ShellyCredential).first() + if not cred: + return None + return ShellyCredentialRead(id=cred.id, server_uri=cred.server_uri) + + +@router.put("/shelly", response_model=ShellyCredentialRead) +async def upsert_shelly_credential( + req: ShellyCredentialUpsert, + db: Session = Depends(get_db), + _: dict = Depends(require_admin), +): + try: + ok = ShellyCloudAPI(req.server_uri, req.auth_key).validate_credentials() + except Exception as exc: + raise HTTPException(502, f"Could not reach Shelly Cloud: {exc}") + if not ok: + raise HTTPException(422, "Shelly Cloud rejected these credentials") + + cred: Optional[ShellyCredential] = db.query(ShellyCredential).first() + if cred: + cred.server_uri = req.server_uri + cred.auth_key_enc = encrypt_secret(req.auth_key) + else: + cred = ShellyCredential( + server_uri=req.server_uri, + auth_key_enc=encrypt_secret(req.auth_key), + ) + db.add(cred) + db.commit() + db.refresh(cred) + return ShellyCredentialRead(id=cred.id, server_uri=cred.server_uri) + + +# ── Mock mode setting ───────────────────────────────────────────────────────── class MockSettingResponse(BaseModel): enabled: bool @@ -76,7 +119,6 @@ async def set_mock_setting( ): cred: Optional[ApiCredential] = db.query(ApiCredential).first() if not cred: - from fastapi import HTTPException raise HTTPException(503, "AVConnect credentials not configured") cred.mock_avconnect = req.enabled db.commit() diff --git a/src/routers/gates.py b/src/routers/gates.py index a799990..4fe7a5c 100644 --- a/src/routers/gates.py +++ b/src/routers/gates.py @@ -8,7 +8,7 @@ from sqlalchemy.orm import Session from core.auth import decrypt_secret from core.config import utcnow -from core.database import ApiCredential, GateAccessLog, GateDB, Keypass, TelegramConfig, get_db +from core.database import ApiCredential, GateAccessLog, GateDB, Keypass, ShellyCredential, TelegramConfig, get_db from core.dependencies import require_admin, require_manager, require_keypass from core.schemas import GateCreate, GatePublicResponse, GateResponse from services.gates import call_open_gate @@ -35,6 +35,49 @@ def _notify(db: Session, gate_name: str, opened_by: str, ip: str | None) -> None ).start() +def _validate_gate_create(req: GateCreate) -> None: + if req.gate_type not in ("car", "pedestrian"): + raise HTTPException(400, "gate_type must be 'car' or 'pedestrian'") + if req.api_provider not in ("avconnect", "shelly"): + raise HTTPException(400, "api_provider must be 'avconnect' or 'shelly'") + if req.api_provider == "avconnect" and not req.avconnect_macro_id: + raise HTTPException(400, "avconnect_macro_id is required for AVConnect gates") + if req.api_provider == "shelly" and not req.shelly_device_id: + raise HTTPException(400, "shelly_device_id is required for Shelly gates") + + +def _do_open_gate( + gate_db: GateDB, + db: Session, + mock: bool, +) -> tuple[bool, Optional[str], Optional[str]]: + """Dispatch the gate-open call to the appropriate API provider.""" + if gate_db.api_provider == "shelly": + shelly_cred: Optional[ShellyCredential] = db.query(ShellyCredential).first() + if not shelly_cred: + raise HTTPException(503, "Shelly credentials not configured") + return call_open_gate( + api_provider="shelly", + shelly_device_id=gate_db.shelly_device_id, + shelly_server_uri=shelly_cred.server_uri, + shelly_auth_key=decrypt_secret(shelly_cred.auth_key_enc), + mock=mock, + ) + + # AVConnect (default) + av_cred: Optional[ApiCredential] = db.query(ApiCredential).first() + if not av_cred: + raise HTTPException(503, "AVConnect credentials not configured") + return call_open_gate( + api_provider="avconnect", + avconnect_macro_id=gate_db.avconnect_macro_id, + avconnect_username=av_cred.username, + avconnect_password=decrypt_secret(av_cred.password_enc), + avconnect_session_id=av_cred.session_id, + mock=mock, + ) + + # ── Admin: gate CRUD ────────────────────────────────────────────────────────── @router.get("/api/admin/gates", response_model=list[GateResponse]) @@ -50,8 +93,7 @@ async def create_gate( db: Session = Depends(get_db), _: dict = Depends(require_admin), ): - if req.gate_type not in ("car", "pedestrian"): - raise HTTPException(400, "gate_type must be 'car' or 'pedestrian'") + _validate_gate_create(req) gate = GateDB(**req.model_dump()) db.add(gate) db.commit() @@ -66,6 +108,7 @@ async def update_gate( db: Session = Depends(get_db), _: dict = Depends(require_admin), ): + _validate_gate_create(req) gate: Optional[GateDB] = db.query(GateDB).filter(GateDB.id == gate_id).first() if not gate: raise HTTPException(404, "Gate not found") @@ -102,21 +145,14 @@ async def admin_open_gate( if gate_db.status != "enabled": raise HTTPException(409, "Gate is disabled") - cred_db: Optional[ApiCredential] = db.query(ApiCredential).first() - if not cred_db: - raise HTTPException(503, "AVConnect credentials not configured") + # Determine mock mode from AVConnect credential record (applies to all providers) + av_cred: Optional[ApiCredential] = db.query(ApiCredential).first() + mock = bool(av_cred.mock_avconnect) if av_cred else False ip = request.client.host if request.client else None ua = request.headers.get("User-Agent") - mock = bool(cred_db.mock_avconnect) - success, error_msg, new_sid = call_open_gate( - gate_db.avconnect_macro_id, - cred_db.username, - decrypt_secret(cred_db.password_enc), - cred_db.session_id, - mock=mock, - ) + success, error_msg, new_sid = _do_open_gate(gate_db, db, mock) db.add(GateAccessLog( timestamp=utcnow(), @@ -130,8 +166,8 @@ async def admin_open_gate( error=error_msg, )) - if new_sid and new_sid != cred_db.session_id: - cred_db.session_id = new_sid + if new_sid and av_cred and new_sid != av_cred.session_id: + av_cred.session_id = new_sid db.commit() if not success: @@ -169,25 +205,17 @@ async def open_gate( if not gate_db: raise HTTPException(404, "Gate not found or disabled") - cred_db: Optional[ApiCredential] = db.query(ApiCredential).first() - if not cred_db: - raise HTTPException(503, "AVConnect credentials not configured") - - ip = request.client.host if request.client else None - ua = request.headers.get("User-Agent") - allowed = json.loads(_kp.allowed_gates) if _kp.allowed_gates else None if allowed is not None and gate_id not in allowed: raise HTTPException(403, "This keypass does not have access to this gate") - mock = bool(cred_db.mock_avconnect) - success, error_msg, new_sid = call_open_gate( - gate_db.avconnect_macro_id, - cred_db.username, - decrypt_secret(cred_db.password_enc), - cred_db.session_id, - mock=mock, - ) + av_cred: Optional[ApiCredential] = db.query(ApiCredential).first() + mock = bool(av_cred.mock_avconnect) if av_cred else False + + ip = request.client.host if request.client else None + ua = request.headers.get("User-Agent") + + success, error_msg, new_sid = _do_open_gate(gate_db, db, mock) db.add(GateAccessLog( timestamp=utcnow(), @@ -201,8 +229,8 @@ async def open_gate( error=error_msg, )) - if new_sid and new_sid != cred_db.session_id: - cred_db.session_id = new_sid + if new_sid and av_cred and new_sid != av_cred.session_id: + av_cred.session_id = new_sid db.commit() if not success: @@ -210,5 +238,5 @@ async def open_gate( raise HTTPException(502, error_msg or "Gate operation failed") logger.info("Gate opened by keypass: gate_id=%d gate=%r keypass=%r ip=%r", gate_db.id, gate_db.name, f"{_kp.description} ({_kp.code})", ip) - _notify(db, gate_db.name, f"{_kp.description} ({_kp.code})", ip) + _notify(db, gate_db.name, f"{_kp.description}", ip) return {"success": True, "gate": gate_db.name} diff --git a/src/services/gates.py b/src/services/gates.py index 23cb049..fb3899d 100644 --- a/src/services/gates.py +++ b/src/services/gates.py @@ -1,21 +1,37 @@ from typing import Optional from .avconnect import AVConnectAPI +from .shelly import ShellyCloudAPI def call_open_gate( - macro_id: str, - username: str, - password: str, - session_id: Optional[str] = None, + api_provider: str, + avconnect_macro_id: Optional[str] = None, + avconnect_username: Optional[str] = None, + avconnect_password: Optional[str] = None, + avconnect_session_id: Optional[str] = None, + shelly_device_id: Optional[str] = None, + shelly_server_uri: Optional[str] = None, + shelly_auth_key: Optional[str] = None, mock: bool = False, ) -> tuple[bool, Optional[str], Optional[str]]: - """Attempt to open a gate. Returns (success, error_msg, new_session_id).""" + """Attempt to open a gate. Returns (success, error_msg, new_avconnect_session_id).""" if mock: return True, None, None + + if api_provider == "shelly": + try: + assert shelly_server_uri and shelly_auth_key and shelly_device_id + ShellyCloudAPI(shelly_server_uri, shelly_auth_key).open_gate(shelly_device_id) + return True, None, None + except Exception as e: + return False, str(e), None + + # Default: AVConnect try: - api = AVConnectAPI(username, password, session_id) - ok, new_sid = api.exec_gate_macro(macro_id) + assert avconnect_username and avconnect_password and avconnect_macro_id + api = AVConnectAPI(avconnect_username, avconnect_password, avconnect_session_id) + ok, new_sid = api.exec_gate_macro(avconnect_macro_id) if not ok: return False, "Gate did not confirm open", new_sid return True, None, new_sid diff --git a/src/static/admin.html b/src/static/admin.html index 8811ea0..35bbf95 100644 --- a/src/static/admin.html +++ b/src/static/admin.html @@ -122,7 +122,7 @@