Add middleware for Gitea webhook signatures

This commit is contained in:
2026-05-01 15:12:16 +02:00
parent d9fa676605
commit 30a982c95d
3 changed files with 83 additions and 1 deletions
+2 -1
View File
@@ -6,4 +6,5 @@
- Add basic API structure
- Add configuration file template
- Add models for Gitea webhooks
- Add models for Gitea webhooks
- Add middleware for Gitea webhook signatures
@@ -0,0 +1,19 @@
# Copyright 2026 jCloud
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
# http://www.apache.org/licenses/LICENSE-2.0
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from . import signature
__all__ = [
'signature'
]
@@ -0,0 +1,62 @@
# Copyright 2026 jCloud
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
# http://www.apache.org/licenses/LICENSE-2.0
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import hmac
import hashlib
from fastapi import Request, HTTPException
from starlette.middleware.base import BaseHTTPMiddleware
__all__ = [
'GiteaSignatureMiddleware'
]
class GiteaSignatureMiddleware(BaseHTTPMiddleware):
'''
A middleware to verify the Gitea signature.
:param app: The (FastAPI) app.
:param secret: The secret.
:type secret: bytes
'''
def __init__(self, app, secret: bytes) -> None:
super().__init__(app)
self.secret = secret
async def dispatch(self, request: Request, call_next):
'''
Dispatch.
:param request: The request.
:type request: Request
'''
if request.url.path.startswith('/webhook/gitea'):
signature = request.headers.get('X-Gitea-Signature')
if not signature:
raise HTTPException(status_code = 401, detail = 'Missing signature')
body = await request.body()
if not hmac.compare_digest(
f'sha256={hmac.new(
self.secret,
body,
hashlib.sha256
).digest()}',
signature
):
raise HTTPException(status_code = 401, detail = 'Invalid signature')
return await call_next(request)