[Python] Push don't arrive, Google Drive API V3

I wrote a script in python, specified everything that was written in the documentation for the API and the library, but notifications do not come. I checked WebHook manually - it works. WebHook has an https:// address, I use tuna for this. I tunnel localhost (127.0. 0.1:8000), the address is accessible from external devices.

import hmac
import hashlib
import ipaddress

from datetime import datetime, timedelta

from uuid_extensions import uuid7

from loguru import logger

from contextlib import asynccontextmanager

from typing import Optional

from fastapi import FastAPI, Request, HTTPException, Header

from data.config import settings

from create_bot import service

ALLOWED_IPS = {"8.8.8.0/24", "2001:4860:4860::/64"} # Диапазоны Google адресов

"""
Вебхук на фастапи, чтобы принимать обновления от гугл диска
"""


async def subscribe_to_drive():
try:
channel = {
"id": str(uuid7()), # Генерируем новый уникальный ID
"type": "web_hook",
"address": "https://cg38qd-213-217-1-130.ru.tuna.am/drive-webhook/",
}

response = service.files().watch(fileId="1dWEzyrnw57UbycVSsUFvp2KCNmC3ILXY", body=channel).execute()
logger.success(f"Новая подписка создана. ID: {channel['id']}")
return response

except Exception as e:
logger.critical(f"Неизвестная ошибка: {e}")

@asynccontextmanager
async def lifespan(app: FastAPI):
await subscribe_to_drive()
yield


app = FastAPI(lifespan=lifespan)


# @app.middleware("http")
# async def verify_google_ip(request: Request, call_next):
# client_ip = request.client.host
# if not any(ipaddress.ip_address(client_ip) in ipaddress.ip_network(net) for net in ALLOWED_IPS):
# raise HTTPException(status_code=403)
# return await call_next(request)


# подпись Google
# async def verify_google_signature(payload: bytes, signature: str) -> bool:
# logger.info("Верификация")
# computed_sig = hmac.new(settings.WEBHOOK_SECRET.encode(), payload, hashlib.sha1).hexdigest()
# return hmac.compare_digest(computed_sig, signature)


# вебхук
@app.post("/drive-webhook/")
async def handle_webhook(
request: Request,
x_goog_resource_state: str = Header(...),
x_goog_resource_id: str = Header(...),
x_goog_signature: Optional[str] = Header(None)
):
logger.info("Прилетел запрос")

# body = await request.body()
# if not await verify_google_signature(body, x_goog_signature):
# raise HTTPException(status_code=403, detail="Invalid signature")

if x_goog_resource_state == "change":
file_id = x_goog_resource_id
logger.info(f"Файл изменён: {file_id}")

return {"status": "ok"}


The code that is in create_bot, and serves to initialize the application

# Google API
creds = service_account.Credentials.from_service_account_file(
"secrets/credentials.json"
)
service = build("drive", "v3", credentials=creds)

Continue reading...