alexei.dolgolyov/notify-bridge
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 31 Wiki Activity

feat: Google Photos provider backend + API hardening

Browse Source 
- Add Google Photos provider: client, models, change detector, capabilities
- Add notification templates (en/ru) for all GP event slots
- Add command templates (en/ru) for GP bot commands
- Register GP in slot/command loaders, capabilities, and seeds
- Harden provider API: validate OAuth credentials on create/update
- Add internal URL rewriting for asset fetches (LAN optimization)
- Fix template renderer to handle missing variables gracefully
- Improve webhook command routing for multi-provider support
- Add provider health check endpoint and watcher improvements
This commit is contained in:
Admin
2026-03-25 22:07:03 +03:00
parent 337276113d
commit 307871cae5
73 changed files with 1154 additions and 144 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/server/src/notify_bridge_server/api/command_template_configs.py
+3 -1
View File
@@ -104,7 +104,9 @@ async def _get(session: AsyncSession, config_id: int, user_id: int) -> CommandTe
# ---------------------------------------------------------------------------
@router.get("/variables")
async def get_command_variables():
async def get_command_variables(
user: User = Depends(get_current_user),
):
"""Get variable reference for each command template slot."""
common_vars = {
"locale": "Current locale (en/ru)",
packages/server/src/notify_bridge_server/api/providers.py
+121 -69
View File
@@ -13,7 +13,7 @@ import aiohttp
from ..auth.dependencies import get_current_user
from ..database.engine import get_session
from ..database.models import ServiceProvider, User
from ..services import make_immich_provider, make_gitea_provider, make_planka_provider, make_nut_provider
from ..services import make_immich_provider, make_gitea_provider, make_planka_provider, make_nut_provider, make_google_photos_provider
_LOGGER = logging.getLogger(__name__)
@@ -76,12 +76,19 @@ class NutProviderConfig(BaseModel):
password: str | None = None
class GooglePhotosProviderConfig(BaseModel):
client_id: str
client_secret: str
refresh_token: str
_PROVIDER_CONFIG_MODELS: dict[str, type[BaseModel]] = {
"immich": ImmichProviderConfig,
"gitea": GiteaProviderConfig,
"planka": PlankaProviderConfig,
"scheduler": SchedulerProviderConfig,
"nut": NutProviderConfig,
"google_photos": GooglePhotosProviderConfig,
}
@@ -122,65 +129,93 @@ async def create_provider(
_validate_provider_config(body.type, body.config)
# Validate connection for known provider types
if body.type == "immich":
from notify_bridge_core.providers.immich import ImmichServiceProvider
config = body.config
async with aiohttp.ClientSession() as http_session:
immich = ImmichServiceProvider(
http_session, config.get("url", ""), config.get("api_key", ""),
config.get("external_domain"), body.name,
)
test_result = await immich.test_connection()
try:
if body.type == "immich":
from notify_bridge_core.providers.immich import ImmichServiceProvider
config = body.config
async with aiohttp.ClientSession() as http_session:
immich = ImmichServiceProvider(
http_session, config.get("url", ""), config.get("api_key", ""),
config.get("external_domain"), body.name,
)
test_result = await immich.test_connection()
if not test_result.get("ok"):
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=test_result.get("message", f"Cannot connect to {body.type} provider"),
)
# Store external_domain from server config if available
if test_result.get("external_domain"):
config["external_domain"] = test_result["external_domain"]
elif body.type == "gitea":
config = body.config
# api_token is optional (webhook_secret is required, but token only for repo listing)
if config.get("api_token"):
async with aiohttp.ClientSession() as http_session:
from notify_bridge_core.providers.gitea import GiteaServiceProvider
gitea = GiteaServiceProvider(
http_session, config.get("url", ""), config.get("api_token", ""), body.name,
)
test_result = await gitea.test_connection()
if not test_result.get("ok"):
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=test_result.get("message", "Cannot connect to Gitea"),
)
elif body.type == "planka":
config = body.config
if config.get("api_key"):
async with aiohttp.ClientSession() as http_session:
from notify_bridge_core.providers.planka import PlankaServiceProvider
planka = PlankaServiceProvider(
http_session, config.get("url", ""), config.get("api_key", ""), body.name,
)
test_result = await planka.test_connection()
if not test_result.get("ok"):
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=test_result.get("message", "Cannot connect to Planka"),
)
elif body.type == "nut":
nut = make_nut_provider(ServiceProvider(
id=0, user_id=0, type="nut", name=body.name, config=body.config,
))
test_result = await nut.test_connection()
if not test_result.get("ok"):
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=test_result.get("message", f"Cannot connect to {body.type} provider"),
detail=test_result.get("message", "Cannot connect to NUT server"),
)
# Store external_domain from server config if available
if test_result.get("external_domain"):
config["external_domain"] = test_result["external_domain"]
elif body.type == "gitea":
config = body.config
# api_token is optional (webhook_secret is required, but token only for repo listing)
if config.get("api_token"):
elif body.type == "google_photos":
config = body.config
async with aiohttp.ClientSession() as http_session:
from notify_bridge_core.providers.gitea import GiteaServiceProvider
gitea = GiteaServiceProvider(
http_session, config.get("url", ""), config.get("api_token", ""), body.name,
from notify_bridge_core.providers.google_photos import GooglePhotosServiceProvider
gp = GooglePhotosServiceProvider(
http_session, config.get("client_id", ""), config.get("client_secret", ""),
config.get("refresh_token", ""), body.name,
)
test_result = await gitea.test_connection()
test_result = await gp.test_connection()
if not test_result.get("ok"):
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=test_result.get("message", "Cannot connect to Gitea"),
detail=test_result.get("message", "Cannot connect to Google Photos"),
)
elif body.type == "planka":
config = body.config
if config.get("api_key"):
async with aiohttp.ClientSession() as http_session:
from notify_bridge_core.providers.planka import PlankaServiceProvider
planka = PlankaServiceProvider(
http_session, config.get("url", ""), config.get("api_key", ""), body.name,
)
test_result = await planka.test_connection()
if not test_result.get("ok"):
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=test_result.get("message", "Cannot connect to Planka"),
)
elif body.type == "nut":
nut = make_nut_provider(ServiceProvider(
id=0, user_id=0, type="nut", name=body.name, config=body.config,
))
test_result = await nut.test_connection()
if not test_result.get("ok"):
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=test_result.get("message", "Cannot connect to NUT server"),
)
except HTTPException:
raise
except aiohttp.ClientError as err:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Connection error: {err}",
)
except OSError as err:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Connection error: {err}",
)
# Scheduler: no validation needed (virtual provider)
@@ -198,7 +233,9 @@ async def create_provider(
@router.get("/capabilities")
async def list_provider_capabilities():
async def list_provider_capabilities(
user: User = Depends(get_current_user),
):
"""List capabilities for all registered provider types."""
from notify_bridge_core.providers.capabilities import get_all_capabilities
result = {}
@@ -218,7 +255,10 @@ async def list_provider_capabilities():
@router.get("/capabilities/{provider_type}")
async def get_provider_capabilities(provider_type: str):
async def get_provider_capabilities(
provider_type: str,
user: User = Depends(get_current_user),
):
"""Get capabilities for a provider type (events, slots, commands)."""
from notify_bridge_core.providers.capabilities import get_capabilities
caps = get_capabilities(provider_type)
@@ -324,6 +364,21 @@ async def update_provider(
status_code=status.HTTP_400_BAD_REQUEST,
detail=test_result.get("message", "Cannot connect to NUT server"),
)
elif config_changed and provider.type == "google_photos":
try:
async with aiohttp.ClientSession() as http_session:
gp = make_google_photos_provider(http_session, provider)
test_result = await gp.test_connection()
if not test_result.get("ok"):
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=test_result.get("message", "Cannot connect to Google Photos"),
)
except aiohttp.ClientError as err:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Connection error: {err}",
)
session.add(provider)
await session.commit()
@@ -380,6 +435,11 @@ async def test_provider(
nut = make_nut_provider(provider)
return await nut.test_connection()
if provider.type == "google_photos":
async with aiohttp.ClientSession() as http_session:
gp = make_google_photos_provider(http_session, provider)
return await gp.test_connection()
return {"ok": False, "message": f"Unknown provider type: {provider.type}"}
@@ -400,22 +460,8 @@ async def list_people(
provider.config.get("url", ""),
provider.config.get("api_key", ""),
)
try:
async with http_session.get(
f"{client.url}/api/people",
headers={"x-api-key": client.api_key},
ssl=False,
) as response:
if response.status == 200:
data = await response.json()
people_list = data.get("people", data) if isinstance(data, dict) else data
return [
{"id": p["id"], "name": p.get("name", "")}
for p in people_list
if p.get("name")
]
except Exception as e:
_LOGGER.error("Failed to fetch people: %s", e)
people = await client.get_people()
return [{"id": pid, "name": name} for pid, name in people.items()]
return []
@@ -452,6 +498,11 @@ async def list_collections(
nut = make_nut_provider(provider)
return await nut.list_collections()
if provider.type == "google_photos":
async with aiohttp.ClientSession() as http_session:
gp = make_google_photos_provider(http_session, provider)
return await gp.list_collections()
return []
@@ -510,10 +561,11 @@ def _provider_response(p: ServiceProvider) -> dict:
"""Build a safe response dict for a provider."""
config = dict(p.config)
# Mask sensitive fields
for secret_field in ("api_key", "api_token", "webhook_secret", "password"):
for secret_field in ("api_key", "api_token", "webhook_secret", "password",
"client_secret", "refresh_token"):
if secret_field in config:
key = config[secret_field]
config[secret_field] = f"{key[:8]}...{key[-4:]}" if len(key) > 12 else "***"
config[secret_field] = f"***{key[-4:]}" if len(key) > 4 else "***"
return {
"id": p.id,
"type": p.type,
packages/server/src/notify_bridge_server/api/targets.py
+26 -18
View File
@@ -79,17 +79,20 @@ async def list_targets(
)
targets = result.all()
# Load receivers for each target
target_receivers: dict[int, list[TargetReceiver]] = {}
for tgt in targets:
# Batch-load all receivers for the user's targets in one query
target_ids = [tgt.id for tgt in targets]
target_receivers: dict[int, list[TargetReceiver]] = {tid: [] for tid in target_ids}
if target_ids:
recv_result = await session.exec(
select(TargetReceiver).where(TargetReceiver.target_id == tgt.id)
select(TargetReceiver).where(TargetReceiver.target_id.in_(target_ids))
)
target_receivers[tgt.id] = list(recv_result.all())
for recv in recv_result.all():
target_receivers[recv.target_id].append(recv)
# Resolve chat names and languages from receivers for telegram targets
# Batch-load chat names and languages for all telegram targets
chat_names: dict[str, str] = {}
chat_languages: dict[str, str] = {}
chat_lookups: list[tuple[int, str]] = [] # (bot_id, chat_id)
for tgt in targets:
if tgt.type == "telegram":
bot_id = tgt.config.get("bot_id")
@@ -98,18 +101,23 @@ async def list_targets(
for recv in target_receivers.get(tgt.id, []):
chat_id = str(recv.config.get("chat_id", ""))
if chat_id:
chat_result = await session.exec(
select(TelegramChat).where(
TelegramChat.bot_id == bot_id,
TelegramChat.chat_id == chat_id,
)
)
chat = chat_result.first()
if chat:
chat_names[f"{bot_id}_{chat_id}"] = chat.title or chat.username or ""
lang = getattr(chat, 'language_override', '') or getattr(chat, 'language_code', '') or ''
if lang:
chat_languages[f"{bot_id}_{chat_id}"] = lang
chat_lookups.append((bot_id, chat_id))
if chat_lookups:
all_bot_ids = list({bl[0] for bl in chat_lookups})
all_chat_ids = list({bl[1] for bl in chat_lookups})
chat_result = await session.exec(
select(TelegramChat).where(
TelegramChat.bot_id.in_(all_bot_ids),
TelegramChat.chat_id.in_(all_chat_ids),
)
)
for chat in chat_result.all():
key = f"{chat.bot_id}_{chat.chat_id}"
chat_names[key] = chat.title or chat.username or ""
lang = getattr(chat, 'language_override', '') or getattr(chat, 'language_code', '') or ''
if lang:
chat_languages[key] = lang
# Build lookup for broadcast child target resolution
target_map = {t.id: t for t in targets}
packages/server/src/notify_bridge_server/api/template_configs.py
+3 -1
View File
@@ -130,7 +130,9 @@ async def list_configs(
@router.get("/variables")
async def get_template_variables():
async def get_template_variables(
user: User = Depends(get_current_user),
):
"""Get template variable reference grouped by slot.
Returns a dict keyed by template slot name, each containing:
packages/server/src/notify_bridge_server/api/template_vars.py
+8 -2
View File
@@ -1,16 +1,22 @@
"""Template variable documentation endpoint."""
from fastapi import APIRouter
from fastapi import APIRouter, Depends
from notify_bridge_core.providers.base import ServiceProviderType
from notify_bridge_core.providers.immich import ImmichServiceProvider # noqa: F401 — triggers registration
from notify_bridge_core.templates.variables import registry
from ..auth.dependencies import get_current_user
from ..database.models import User
router = APIRouter(prefix="/api/template-vars", tags=["template-vars"])
@router.get("")
async def get_template_variables(provider_type: str | None = None):
async def get_template_variables(
provider_type: str | None = None,
user: User = Depends(get_current_user),
):
"""Get available template variables, optionally filtered by provider type."""
if provider_type:
try:
packages/server/src/notify_bridge_server/api/users.py
+5 -2
View File
@@ -55,6 +55,9 @@ async def create_user(
if result.first():
raise HTTPException(status_code=409, detail="Username already exists")
if len(body.password) < 8:
raise HTTPException(status_code=400, detail="Password must be at least 8 characters")
user = User(
username=body.username,
hashed_password=bcrypt.hashpw(body.password.encode(), bcrypt.gensalt()).decode(),
@@ -81,8 +84,8 @@ async def reset_user_password(
user = await session.get(User, user_id)
if not user:
raise HTTPException(status_code=404, detail="User not found")
if len(body.new_password) < 6:
raise HTTPException(status_code=400, detail="Password must be at least 6 characters")
if len(body.new_password) < 8:
raise HTTPException(status_code=400, detail="Password must be at least 8 characters")
user.hashed_password = bcrypt.hashpw(body.new_password.encode(), bcrypt.gensalt()).decode()
session.add(user)
await session.commit()
packages/server/src/notify_bridge_server/api/webhooks.py
+4 -2
View File
@@ -156,7 +156,6 @@ async def gitea_webhook(provider_id: int, request: Request):
)},
},
))
await session.commit()
# Dispatch to targets
dispatcher = NotificationDispatcher()
@@ -172,6 +171,8 @@ async def gitea_webhook(provider_id: int, request: Request):
tracker.id, r.get("error", "unknown"),
)
await session.commit()
return {"ok": True, "dispatched": dispatched}
@@ -268,7 +269,6 @@ async def planka_webhook(provider_id: int, request: Request):
)},
},
))
await session.commit()
# Dispatch to targets
dispatcher = NotificationDispatcher()
@@ -284,6 +284,8 @@ async def planka_webhook(provider_id: int, request: Request):
tracker.id, r.get("error", "unknown"),
)
await session.commit()
return {"ok": True, "dispatched": dispatched}
packages/server/src/notify_bridge_server/commands/webhook.py
+2 -1
View File
@@ -2,6 +2,7 @@
from __future__ import annotations
import hmac
import logging
from typing import Any
@@ -40,7 +41,7 @@ async def telegram_webhook(
"""Handle incoming Telegram messages — route commands to handlers."""
# Validate webhook secret if configured
if _webhook_secret:
if x_telegram_bot_api_secret_token != _webhook_secret:
if not hmac.compare_digest(x_telegram_bot_api_secret_token or "", _webhook_secret):
raise HTTPException(status_code=403, detail="Invalid webhook secret")
# Find bot by opaque webhook path ID (not by token — token must not appear in URLs)
packages/server/src/notify_bridge_server/config.py
+2 -2
View File
@@ -33,8 +33,8 @@ class Settings(BaseSettings):
telegram_webhook_secret: str = ""
cors_allowed_origins: str = "*"
"""Comma-separated allowed origins for CORS (e.g. 'http://localhost:5173,https://myapp.com'). Use '*' for dev."""
cors_allowed_origins: str = "http://localhost:5173"
"""Comma-separated allowed origins for CORS (e.g. 'http://localhost:5173,https://myapp.com')."""
static_dir: str = ""
"""Path to frontend static files. Set to serve SvelteKit build via FastAPI (e.g. /app/static in Docker)."""
packages/server/src/notify_bridge_server/database/seeds.py
+14
View File
@@ -152,6 +152,7 @@ async def _seed_default_templates() -> None:
await _seed_provider_template(session, "planka", "Planka")
await _seed_provider_template(session, "scheduler", "Scheduler")
await _seed_provider_template(session, "nut", "NUT")
await _seed_provider_template(session, "google_photos", "Google Photos")
await session.commit()
@@ -175,6 +176,9 @@ async def _seed_default_command_templates() -> None:
await _seed_provider_command_template(
session, "nut", "Default NUT Commands", "Default NUT command templates",
)
await _seed_provider_command_template(
session, "google_photos", "Default Google Photos Commands", "Default Google Photos command templates",
)
await session.commit()
@@ -305,6 +309,16 @@ async def _seed_default_command_configs() -> None:
"default_count": 5,
"rate_limits": {"api": 15, "default": 10},
},
{
"provider_type": "google_photos",
"name": "Default Google Photos",
"enabled_commands": [
"help", "status", "albums", "latest", "search", "random",
],
"response_mode": "media",
"default_count": 5,
"rate_limits": {"search": 30, "default": 10},
},
]
for cfg in defaults:
packages/server/src/notify_bridge_server/main.py
+23 -2
View File
@@ -11,8 +11,11 @@ from slowapi.middleware import SlowAPIMiddleware
# Ensure app-level loggers are visible
logging.basicConfig(level=logging.INFO)
logging.getLogger("notify_bridge_server").setLevel(logging.DEBUG)
logging.getLogger("notify_bridge_core").setLevel(logging.DEBUG)
from .config import settings as _log_cfg
_log_level = logging.DEBUG if _log_cfg.debug else logging.INFO
logging.getLogger("notify_bridge_server").setLevel(_log_level)
logging.getLogger("notify_bridge_core").setLevel(_log_level)
from .database.engine import init_db
from .database.models import * # noqa: F401,F403 — ensure all models registered
@@ -77,6 +80,24 @@ async def lifespan(app: FastAPI):
app = FastAPI(title="Notify Bridge", version="0.1.0", lifespan=lifespan)
# --- Security headers ---
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request as StarletteRequest
from starlette.responses import Response as StarletteResponse
class SecurityHeadersMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: StarletteRequest, call_next):
response: StarletteResponse = await call_next(request)
response.headers["X-Content-Type-Options"] = "nosniff"
response.headers["X-Frame-Options"] = "DENY"
response.headers["X-XSS-Protection"] = "1; mode=block"
response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"
return response
app.add_middleware(SecurityHeadersMiddleware)
# --- Rate limiting ---
from .auth.routes import limiter
app.state.limiter = limiter
packages/server/src/notify_bridge_server/services/__init__.py
+13
View File
@@ -4,6 +4,7 @@ from notify_bridge_core.providers.immich import ImmichServiceProvider
from notify_bridge_core.providers.gitea import GiteaServiceProvider
from notify_bridge_core.providers.planka import PlankaServiceProvider
from notify_bridge_core.providers.nut import NutServiceProvider
from notify_bridge_core.providers.google_photos import GooglePhotosServiceProvider
from ..database.models import ServiceProvider
@@ -52,3 +53,15 @@ def make_nut_provider(provider: ServiceProvider) -> NutServiceProvider:
password=config.get("password"),
name=provider.name,
)
def make_google_photos_provider(http_session, provider: ServiceProvider) -> GooglePhotosServiceProvider:
"""Create a GooglePhotosServiceProvider from a DB provider model."""
config = provider.config or {}
return GooglePhotosServiceProvider(
http_session,
config.get("client_id", ""),
config.get("client_secret", ""),
config.get("refresh_token", ""),
provider.name,
)
packages/server/src/notify_bridge_server/services/watcher.py
+34 -12
View File
@@ -2,6 +2,7 @@
from __future__ import annotations
import asyncio
import logging
from typing import Any
@@ -28,6 +29,7 @@ _LOGGER = logging.getLogger(__name__)
# Module-level Telegram file caches — shared across dispatches for reuse
_url_cache: TelegramFileCache | None = None
_asset_cache: TelegramFileCache | None = None
_cache_lock = asyncio.Lock()
async def _get_telegram_caches() -> tuple[TelegramFileCache | None, TelegramFileCache | None]:
@@ -35,18 +37,24 @@ async def _get_telegram_caches() -> tuple[TelegramFileCache | None, TelegramFile
global _url_cache, _asset_cache
if _url_cache is not None:
return _url_cache, _asset_cache
import os
from pathlib import Path
data_dir = os.environ.get("NOTIFY_BRIDGE_DATA_DIR")
if not data_dir:
return None, None
cache_dir = Path(data_dir) / "cache"
_url_cache = TelegramFileCache(JsonFileBackend(cache_dir / "telegram_url_cache.json"))
_asset_cache = TelegramFileCache(JsonFileBackend(cache_dir / "telegram_asset_cache.json"))
await _url_cache.async_load()
await _asset_cache.async_load()
_LOGGER.info("Initialized Telegram file caches in %s", cache_dir)
return _url_cache, _asset_cache
async with _cache_lock:
# Double-check after acquiring lock
if _url_cache is not None:
return _url_cache, _asset_cache
import os
from pathlib import Path
data_dir = os.environ.get("NOTIFY_BRIDGE_DATA_DIR")
if not data_dir:
return None, None
cache_dir = Path(data_dir) / "cache"
url_cache = TelegramFileCache(JsonFileBackend(cache_dir / "telegram_url_cache.json"))
asset_cache = TelegramFileCache(JsonFileBackend(cache_dir / "telegram_asset_cache.json"))
await url_cache.async_load()
await asset_cache.async_load()
_url_cache = url_cache
_asset_cache = asset_cache
_LOGGER.info("Initialized Telegram file caches in %s", cache_dir)
return _url_cache, _asset_cache
async def check_tracker(tracker_id: int) -> dict[str, Any]:
@@ -133,6 +141,20 @@ async def check_tracker(tracker_id: int) -> dict[str, Any]:
name=provider_name,
)
events, new_state = await nut.poll(collection_ids, state_dict)
elif provider_type == "google_photos":
from notify_bridge_core.providers.google_photos import GooglePhotosServiceProvider
async with aiohttp.ClientSession() as http_session:
gp = GooglePhotosServiceProvider(
http_session,
provider_config.get("client_id", ""),
provider_config.get("client_secret", ""),
provider_config.get("refresh_token", ""),
provider_name,
)
connected = await gp.connect()
if not connected:
return {"status": "error", "reason": "failed to connect to Google Photos"}
events, new_state = await gp.poll(collection_ids, state_dict)
else:
return {"status": "error", "reason": f"unsupported provider type: {provider_type}"}