30 Commits

Author SHA1 Message Date
jakob.scheid 7c6ca3b5a4 Add Gitea API router 2026-05-12 18:56:34 +02:00
jakob.scheid b8dc240a8a Add support for disabling the Gitea webhook secret 2026-05-11 22:53:12 +02:00
jakob.scheid 547edfaef0 Generalize the term 'Gitea webhook configuration' to 'Gitea configuration' 2026-05-11 22:46:37 +02:00
jakob.scheid 7ca1d4c609 Update __all__ in integrations.gitea.models.release 2026-05-08 17:45:39 +02:00
jakob.scheid 791e1d972b Add check_enabled in __init__.py 2026-05-08 17:44:43 +02:00
jakob.scheid 0e23b4d349 Bug fix: Remove trailing slash in '/gitea' 2026-05-08 17:42:27 +02:00
jakob.scheid c26291bb1c Add unit tests for integrations.gitea.middlewares.check_enabled.GiteaCheckEnabledMiddleware 2026-05-08 17:41:49 +02:00
jakob.scheid 97f1a802fa Move make_request function to a separate utils module 2026-05-08 16:58:35 +02:00
jakob.scheid 16bd8b25f2 Add middleware to check whether Gitea is enabled 2026-05-08 16:50:18 +02:00
jakob.scheid 944c41095d Add Gitea configuration processing in api.config 2026-05-06 15:33:32 +02:00
jakob.scheid 6afa9af3b2 Add configuration for Gitea integration 2026-05-06 15:32:40 +02:00
jakob.scheid c9e317f7ca Update Gitea webhook path 2026-05-06 01:39:46 +02:00
jakob.scheid 2da8da9f26 Add error logging for integrations.gitea.middlewares.signature.GiteaSignatureMiddleware 2026-05-06 01:35:59 +02:00
jakob.scheid 2db98caaef Bug fix: Set disable_existing_loggers to False 2026-05-06 01:05:07 +02:00
jakob.scheid 4cdd52059d Create logging target option and fallbacks 2026-05-06 00:38:56 +02:00
jakob.scheid b295738c0e Add configuration options for logging 2026-05-06 00:37:41 +02:00
jakob.scheid c0cec78b03 Add API exceptions module 2026-05-06 00:37:05 +02:00
jakob.scheid 81f79783b6 Remove debug line 2026-05-04 20:03:12 +02:00
jakob.scheid d4e7d983fa Remove unnecessary import 2026-05-04 20:02:06 +02:00
jakob.scheid 044ccba66a Divide API into modules 2026-05-04 20:01:43 +02:00
jakob.scheid 10a25b82af Bug fix: HTTPException should not be used in a middleware, so use JSONResponse instead 2026-05-03 19:04:59 +02:00
jakob.scheid 374dc89c09 Add tests with invalid signatures for integrations.gitea.middlewares.signature.GiteaSignatureMiddleware 2026-05-03 18:34:20 +02:00
jakob.scheid 157d054854 Remove unnecessary field expected_exception at the tests for integrations.gitea.middlewares.signature.GiteaSignatureMiddleware 2026-05-03 18:14:50 +02:00
jakob.scheid 58700f2390 Add tests for integrations.gitea.middlewares.signature.GiteaSignatureMiddleware 2026-05-03 18:09:17 +02:00
jakob.scheid 23faf74092 Bug fix: use heximal SHA256 hash instead of bytes (repr)
integrations.gitea.middlewares.signature.GiteaSignatureMiddleware.dispatch
used the Python bytes representation of the SHA256 hash for comparision
in the f-string. Now, the hexadecimal hash is used.
2026-05-03 18:06:45 +02:00
jakob.scheid 30a982c95d Add middleware for Gitea webhook signatures 2026-05-01 15:12:16 +02:00
jakob.scheid d9fa676605 Update package structure 2026-05-01 13:40:51 +02:00
jakob.scheid 3d0dc70db5 Update licenses 2026-05-01 13:23:16 +02:00
jakob.scheid 6ed50eb102 Add models for Gitea webhooks 2026-04-29 20:11:55 +02:00
jakob.scheid 1b26f5c86f Add models for Gitea webhooks 2026-04-29 19:17:58 +02:00
22 changed files with 1346 additions and 146 deletions
+1 -1
View File
@@ -1,4 +1,4 @@
Copyright 2026 jCloud Services GbR Copyright 2026 jCloud
Licensed under the Apache License, Version 2.0 (the "License"); Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License. you may not use this file except in compliance with the License.
+23 -2
View File
@@ -23,8 +23,29 @@
# openapi= # openapi=
[logging] [logging]
# The file for the logs. Default is stdout ('/proc/self/fd/1'). # The target for logging. Possible values are `file` and `stdout`.
# Default is `stdout`. If the value is invalid, the default value is used
# instead if fail_strict is not true. If it is true and the value is
# invalid, it fails.
# logtarget=
# If one of 'true', 'yes', 't', or 'y', stdout is used as a fallback if
# the logging target is a file and the log file cannot be used for
# various reasons. Otherwise, the default value for the property
# logtarget is used as a fallback. Default is false.
# fail_strict=
# The file for the logs. Default is empty.
# logfile= # logfile=
# The log level. Default is INFO. # The log level. Default is INFO.
# loglevel= # loglevel=
[gitea]
# Whether Gitea is enabled ('true', 'yes', 't', or 'y'). Default is
# false.
# enabled=
# The file for the Gitea webhook secret. Leave it empty to disable the
# secret (warning: very insecure).
# webhook_secret_file=
+8 -2
View File
@@ -11,5 +11,11 @@ redoc=/redoc
openapi=/openapi.json openapi=/openapi.json
[logging] [logging]
logfile=/proc/self/fd/1 logtarget=stdout
loglevel=INFO logfile=
fail_strict=false
loglevel=INFO
[gitea]
enabled=false
webhook_secret_file=
+3 -1
View File
@@ -5,4 +5,6 @@
### Added ### Added
- Add basic API structure - Add basic API structure
- Add configuration file template - Add configuration file template
- Add models for Gitea webhooks
- Add middleware for Gitea webhook signatures
+3 -140
View File
@@ -1,4 +1,4 @@
# Copyright 2026 jCloud Services GbR # Copyright 2026 jCloud
# Licensed under the Apache License, Version 2.0 (the "License"); # Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License. # you may not use this file except in compliance with the License.
@@ -12,144 +12,7 @@
# See the License for the specific language governing permissions and # See the License for the specific language governing permissions and
# limitations under the License. # limitations under the License.
from fastapi import FastAPI from .api import main
import os
import uvicorn
import argparse
import ipaddress
import jcloud_config_parser
import logging
import pathlib
def existing_file(path: str) -> pathlib.Path:
'''
Checks whether a path is a file and returns it if it exists.
:param path: The path
:type path: str
:raises argparse.ArgumentTypeError: If the path is a directory or does not exist.
:return: The path
:rtype: pathlib.Path
'''
if os.path.isdir(path):
raise argparse.ArgumentTypeError(f'\'{path}\': Is a directory')
if not os.path.isfile(path):
raise argparse.ArgumentTypeError(f'\'{path}\': No such file or directory')
return pathlib.Path(path)
# parse arguments
argument_parser = argparse.ArgumentParser(description='The jCloud deployment server.')
argument_parser.add_argument('-c', '--config', type=existing_file, default=pathlib.Path(os.path.dirname(__file__) + '/../../config/'), help='The directory containing the configuration files.')
argument_parser.add_argument('-H', '--host', type=ipaddress.ip_address, default=None, help='The host to listen')
argument_parser.add_argument('-p', '--port', type=int, default=None, help='The port to listen')
args = argument_parser.parse_args()
# load default configuration
with open(os.path.dirname(__file__) + '/../../default_configuration.conf') as f:
default_configuration = jcloud_config_parser.ini.INIConfiguration.from_string(f.read())
f.close()
# load configuration
configuration = jcloud_config_parser.ini.INIConfiguration.from_string(
(args.config / 'server.conf').read_text(),
default=default_configuration,
ignore_errors=True
)
logfile = configuration.logging.logfile
if os.path.isdir(logfile):
logfile = None
if not logfile:
logfile = None
LOGLEVEL = logging._nameToLevel.get(configuration.logging.loglevel.upper(), logging._nameToLevel.get(default_configuration.logging.loglevel.upper(), logging.INFO))
# set up logging
LOGGING_FORMAT = '%(asctime)s [%(levelname)s] %(name)s: %(message)s'
logging.basicConfig(level=LOGLEVEL, format=LOGGING_FORMAT)
logger = logging.getLogger('api')
logger.propagate = False
logger.handlers = list()
formatter = logging.Formatter(LOGGING_FORMAT)
if logfile is not None:
logger_file_handler = logging.FileHandler(logfile)
logger_file_handler.setFormatter(formatter)
logger.addHandler(logger_file_handler)
def validate_host(host: str) -> bool:
'''
Checks whether the address is a valid address (``ip:port``), e. g. ``0.0.0.0:3000``.
:param addr: The address
:type addr: str
:return: ``True`` if the address is valid, otherwise ``False``.
:rtype: bool
'''
try:
ipaddress.ip_address(host)
return True
except Exception:
return False
# host
host = args.host or configuration.server.host
if not validate_host(host):
logger.error(f'Error in configuration: [server]host: \'{host}\' is not a valid host, using default value.')
host = default_configuration.server.host
# port
port = args.port or configuration.server.port
if not port.isdigit():
logger.error(f'Error in configuration: [server]port: \'{port}\' is not an integer, using default value.')
port = default_configuration.server.port
else:
if not (0 <= int(port) <= 65535):
logger.error(f'Error in configuration: [server]port: {port} is not between 0 and 65535, using default value.')
port = default_configuration.server.port
port = int(port)
# Initialize FastAPI
app = FastAPI(
docs_url=configuration.docs.swagger,
redoc_url=configuration.docs.redoc,
openapi_url=configuration.docs.openapi
)
if __name__ == '__main__': if __name__ == '__main__':
uvicorn.run( main.main()
app,
host = host,
port = port,
server_header = False,
log_config = {
'version': 1,
'disable_existing_loggers': True,
'handlers': {
'uvicorn_handler': {
'class': 'logging.FileHandler',
'filename': logfile,
'formatter': 'uvicorn_formatter',
},
},
'formatters': {
'uvicorn_formatter': {
'format': LOGGING_FORMAT
},
},
'loggers': {
'uvicorn': {
'handlers': ['uvicorn_handler'],
'level': LOGLEVEL,
'propagate': False
},
},
}
)
@@ -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 argparse
import os
import pathlib
import ipaddress
__all__ = [
'parse_args'
]
def existing_file(path: str) -> pathlib.Path:
'''
Checks whether a path is a file and returns it if it exists.
:param path: The path
:type path: str
:raises argparse.ArgumentTypeError: If the path is a directory or does not exist.
:return: The path
:rtype: pathlib.Path
'''
if os.path.isdir(path):
raise argparse.ArgumentTypeError(f'\'{path}\': Is a directory')
if not os.path.isfile(path):
raise argparse.ArgumentTypeError(f'\'{path}\': No such file or directory')
return pathlib.Path(path)
def parse_args(argv: list[str]) -> argparse.Namespace:
'''
Parses the arguments.
:param argv: The arguments.
:type argv: list[str]
:return: The argument namespace.
:rtype: argparse.Namespace
'''
# parse arguments
argument_parser = argparse.ArgumentParser(description='The jCloud deployment server.')
argument_parser.add_argument('-c', '--config', type=existing_file, default=pathlib.Path(os.path.dirname(__file__) + '/../../../config/'), help='The directory containing the configuration files.')
argument_parser.add_argument('-H', '--host', type=ipaddress.ip_address, default=None, help='The host to listen')
argument_parser.add_argument('-p', '--port', type=int, default=None, help='The port to listen')
return argument_parser.parse_args(argv)
+175
View File
@@ -0,0 +1,175 @@
# 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 dataclasses import dataclass
from .exceptions import Fail
import os
import jcloud_config_parser
import ipaddress
import pathlib
import logging
import argparse
from typing import Optional
__all__ = [
'load_config',
'process_host_and_port',
'GiteaConfig',
'process_gitea_config',
]
def load_config(config_directory: pathlib.Path) -> tuple[
jcloud_config_parser.ini.INIConfiguration,
jcloud_config_parser.ini.INIConfiguration
]:
'''
Loads the configuration and the default configuration.
:param config_directory: The configuration directory.
:type config_directory: pathlib.Path
:return: The configuration.
:rtype: tuple[jcloud_config_parser.ini.INIConfiguration, jcloud_config_parser.ini.INIConfiguration]
'''
# load default configuration
with open(os.path.dirname(__file__) + '/../../../default_configuration.conf') as f:
default_configuration = jcloud_config_parser.ini.INIConfiguration.from_string(f.read())
f.close()
# load configuration
configuration = jcloud_config_parser.ini.INIConfiguration.from_string(
(config_directory / 'server.conf').read_text(),
default=default_configuration,
ignore_errors=True
)
return configuration, default_configuration
def _validate_host(host: str) -> bool:
'''
Checks whether the address is a valid address (``ip:port``), e. g. ``0.0.0.0:3000``.
:param addr: The address
:type addr: str
:return: ``True`` if the address is valid, otherwise ``False``.
:rtype: bool
'''
try:
ipaddress.ip_address(host)
return True
except Exception:
return False
def process_host_and_port(
configuration: jcloud_config_parser.ini.INIConfiguration,
default_configuration: jcloud_config_parser.ini.INIConfiguration,
args: argparse.Namespace,
logger: logging.Logger
) -> tuple[str, int]:
'''
Processes the host and port in the configuration.
Uses default values and the configuration data passed via
command-line arguments as a fallback.
:param configuration: The configuration.
:type configuration: jcloud_config_parser.ini.INIConfiguration
:param default_configuration: The default configuration.
:type default_configuration: jcloud_config_parser.ini.INIConfiguration
:param args: The arguments.
:type args: argparse.Namespace
:return: The host and the port
:rtype: tuple[str, int]
'''
# host
host = args.host or configuration.server.host
if not _validate_host(host):
logger.error(f'Error in configuration: [server]host: \'{host}\' is not a valid host, using default value.')
host = default_configuration.server.host
# port
port = args.port or configuration.server.port
if not port.isdigit():
logger.error(f'Error in configuration: [server]port: \'{port}\' is not an integer, using default value.')
port = default_configuration.server.port
else:
if not (0 <= int(port) <= 65535):
logger.error(f'Error in configuration: [server]port: {port} is not between 0 and 65535, using default value.')
port = default_configuration.server.port
return host, int(port)
@dataclass
class GiteaConfig:
enabled: bool
webhook_secret_file_path: Optional[pathlib.Path]
def _is_readable_file(path: pathlib.Path) -> bool:
'''
Returns whether the file is readable and exists.
:param path: The file path.
:type path: pathlib.Path
:return: Whether the file is readable and exists.
:rtype: bool
'''
if not path.exists():
return False
try:
with open(str(path), 'rb'):
pass
except (OSError, PermissionError):
return False
return True
def process_gitea_config(
configuration: jcloud_config_parser.ini.INIConfiguration,
logger: logging.Logger
) -> GiteaConfig:
'''
Processes the Gitea configuration.
:param configuration: The configuration.
:type configuration: jcloud_config_parser.ini.INIConfiguration
:param logger: The logger.
:type logger: logging.Logger
:return: The Gitea configuration.
:rtype: GiteaConfig
'''
if configuration['gitea'].enabled not in ('true', 'yes', 't', 'y'):
return GiteaConfig(False, None)
secret_file_path = configuration['gitea'].webhook_secret_file
if not secret_file_path: # disable secret
secret_file_path = None
else:
secret_file_path = pathlib.Path(secret_file_path)
if not _is_readable_file(secret_file_path):
logger.critical(f'{secret_file_path}: Cannot read Gitea webhook secret file')
raise Fail(f'{secret_file_path}: Cannot read Gitea webhook secret file', exit_code = 2)
return GiteaConfig(True, secret_file_path)
@@ -0,0 +1,29 @@
# 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.
__all__ = [
'Fail'
]
class Fail(Exception):
'''
The program fails.
:param exit_code: The exit code.
:type exit_code: int
'''
def __init__(self, *args, exit_code: int = 1):
super().__init__()
self.exit_code = exit_code
+147
View File
@@ -0,0 +1,147 @@
# 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 os
import jcloud_config_parser
import logging
import pathlib
from dataclasses import dataclass
from .exceptions import Fail
__all__ = [
'LoggingInfo',
'setup_logging'
]
@dataclass
class LoggingInfo:
logger: logging.Logger
uvicorn_config: dict
def _logfile_fallback_used(path: pathlib.Path) -> bool:
'''
Returns whether the fallback value for the logfile should be used.
:param path: The path to the logfile.
:type path: pathlib.Path
:return: Whether the fallback value should be used.
:rtype: bool
'''
# Path exists and is a directory -> invalid
if path.exists() and path.is_dir():
return True
parent = path.parent or pathlib.Path('.')
# Parent directory missing -> try to create it
if not parent.exists():
try:
parent.mkdir(parents = True, exist_ok = True)
except OSError:
return True
# Check whether the path is writable.
try:
with open(path, 'a'):
pass
except (OSError, PermissionError):
return True
return False
def setup_logging(configuration: jcloud_config_parser.ini.INIConfiguration) -> LoggingInfo:
'''
Sets up logging.
:param configuration: The configuration.
:type configuration: jcloud_config_parser.ini.INIConfiguration
:return: The logger info.
:rtype: LoggingInfo
'''
fail_strict = configuration.logging.fail_strict.lower() in ('true', 'yes', 't', 'y')
logtarget = configuration.logging.logtarget
if logtarget not in ('file', 'stdout'):
if fail_strict:
raise Fail(f'\'{logtarget}\': invalid value for [logging].logtarget. Expected either \'file\' or \'stdout\'.')
logtarget = 'stdout'
if logtarget == 'file':
logfile = pathlib.Path(configuration.logging.logfile)
if _logfile_fallback_used(logfile):
if fail_strict:
raise Fail(f'\'{logfile}\': Cannot use as a log file.')
logtarget = 'stdout' # fallback to stdout
else:
logfile = None
loglevel = logging._nameToLevel.get(
configuration.logging.loglevel.upper(),
logging.INFO
)
# set up logging
logging_format = '%(asctime)s [%(levelname)s] %(name)s: %(message)s'
logging.basicConfig(level=loglevel, format=logging_format)
logger = logging.getLogger('api')
logger.propagate = False
formatter = logging.Formatter(logging_format)
if logtarget == 'file':
logger.handlers = list()
handler = logging.FileHandler(logfile)
handler.setFormatter(formatter)
else:
handler = logging.StreamHandler()
handler.setFormatter(formatter)
logger.addHandler(handler)
uvicorn_config = {
'version': 1,
'disable_existing_loggers': False,
'formatters': {
'uvicorn_formatter': {
'format': logging_format
}
},
'loggers': {
'uvicorn': {
'handlers': ['uvicorn_handler'],
'level': loglevel,
'propagate': False
},
},
'handlers': {
'uvicorn_handler': {
'class': 'logging.FileHandler',
'filename': logfile,
'formatter': 'uvicorn_formatter'
} if logtarget == 'file'
else {
'class': 'logging.StreamHandler',
'formatter': 'uvicorn_formatter'
}
}
}
return LoggingInfo(
logger,
uvicorn_config
)
+72
View File
@@ -0,0 +1,72 @@
# 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 fastapi import FastAPI, Header
import os
import uvicorn
import ipaddress
import jcloud_config_parser
import logging
from .arguments import parse_args
from .config import load_config, process_host_and_port, process_gitea_config
from .logging import setup_logging
from ..integrations.gitea.middlewares.signature import GiteaSignatureMiddleware
from ..integrations.gitea.middlewares.check_enabled import GiteaCheckEnabledMiddleware
from ..integrations.gitea.api.webhooks import make_router
import sys
def main():
args = parse_args(sys.argv[1:])
configuration, default_configuration = load_config(args.config)
logger_info = setup_logging(configuration)
host, port = process_host_and_port(
configuration,
default_configuration,
args,
logger_info.logger
)
gitea_config = process_gitea_config(configuration, logger_info.logger)
# Initialize FastAPI
app = FastAPI(
docs_url=configuration.docs.swagger,
redoc_url=configuration.docs.redoc,
openapi_url=configuration.docs.openapi
)
if gitea_config.webhook_secret_file_path is not None:
app.add_middleware(
GiteaSignatureMiddleware,
secret = gitea_config.webhook_secret_file_path.read_bytes(),
logger = logger_info.logger
) # empty secret is an example value!
app.add_middleware(
GiteaCheckEnabledMiddleware,
configuration = configuration,
logger = logger_info.logger
)
app.include_router(make_router(gitea_config.webhook_secret_file_path is not None))
uvicorn.run(
app,
host = host,
port = port,
server_header = False,
log_config = logger_info.uvicorn_config
)
@@ -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 gitea
__all__ = [
'gitea'
]
@@ -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 models
__all__ = [
'models'
]
@@ -0,0 +1,47 @@
# 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 fastapi import APIRouter, Header
from ..models.release import Release
def make_router(secret_enabled: bool = False) -> APIRouter:
'''
Returns the Gitea webhook router.
:param secret_enabled: Whether the secret is enabled.
:type secret_enabled: bool
:return: The router.
:rtype: APIRouter
'''
router = APIRouter()
if secret_enabled:
x_gitea_signature_header_default = ...
else:
x_gitea_signature_header_default = None
@router.post('/gitea/webhook')
async def gitea_webhook(release: Release, x_gitea_signature = Header(x_gitea_signature_header_default)):
'''
Processes a Gitea webhook.
Works only if Gitea is enabled.
'''
raise NotImplementedError()
return router
@@ -0,0 +1,20 @@
# 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 check_enabled, signature
__all__ = [
'check_enabled',
'signature'
]
@@ -0,0 +1,60 @@
# 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 logging
import jcloud_config_parser
from typing import Optional
from fastapi import FastAPI, Request
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import JSONResponse
__all__ = [
'GiteaCheckEnabledMiddleware'
]
class GiteaCheckEnabledMiddleware(BaseHTTPMiddleware):
'''
A middleware to check whether Gitea webhooks are enabled.
:param app: The (FastAPI) app.
:param configuration: The configuration.
:type configuration: jcloud_config_parser.ini.INIConfiguration
:param logger: The logger.
:type logger: Optional[logging.Logger]
'''
def __init__(self, app, configuration: jcloud_config_parser.ini.INIConfiguration, logger: Optional[logging.Logger] = None) -> None:
super().__init__(app)
self.fastapi_app = app
self.configuration = configuration
self.logger = logger
async def dispatch(self, request: Request, call_next):
'''
Dispatch.
:param request: The request.
:type request: Request
'''
if request.url.path.startswith('/gitea'):
if self.configuration.gitea.enabled not in ('true', 'yes', 't', 'y'):
if self.logger is not None:
self.logger.error(f'{request.url.path}: Gitea is disabled')
return JSONResponse(
status_code = 403,
content = {'detail': 'Gitea disabled'}
)
return await call_next(request)
@@ -0,0 +1,110 @@
# 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
import logging
from typing import Optional
from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import JSONResponse
__all__ = [
'GiteaSignatureMiddleware'
]
def _get_client_host(request: Request, proxy: bool = False, proxy_host_header: Optional[str] = 'X-Forwarded-For') -> str:
'''
Returns the client host.
If ``proxy`` is ``True``, the header that is specified in
``proxy_host_header`` is used to get the client host. Otherwise, the
real client host is returned.
:param request: The request object.
:type request: Request
:param proxy: Whether a proxy is used.
:type proxy: bool
:param proxy_host_header: The header set by the proxy containing the
original client host.
:type proxy_host_header: Optional[str]
:raises ValueError: If ``proxy`` is ``True`` and
``proxy_host_header`` is ``None``.
:return: The client host.
:rtype: str
'''
if proxy:
if not proxy_host_header:
raise ValueError('Passing a proxy host header is necessary if proxy is True')
return request.headers.get(proxy_host_header)
else:
return request.client.host
class GiteaSignatureMiddleware(BaseHTTPMiddleware):
'''
A middleware to verify the Gitea signature.
:param app: The (FastAPI) app.
:param secret: The secret.
:type secret: bytes
:param logger: The logger.
:type logger: Optional[logging.Logger]
'''
def __init__(self, app, secret: bytes, logger: Optional[logging.Logger] = None) -> None:
super().__init__(app)
self.secret = secret
self.logger = logger
async def dispatch(self, request: Request, call_next):
'''
Dispatch.
:param request: The request.
:type request: Request
'''
if request.url.path.startswith('/gitea/webhook'):
signature = request.headers.get('X-Gitea-Signature')
if not signature:
if self.logger is not None:
self.logger.error(f'Permission denied: no signature (client: {_get_client_host(request)})')
return JSONResponse(
status_code = 401,
content = {'detail': 'Invalid signature'}
)
body = await request.body()
if not hmac.compare_digest(
f'sha256={hmac.new(
self.secret,
body,
hashlib.sha256
).hexdigest()}',
signature
):
if self.logger is not None:
self.logger.error(f'Permission denied: invalid signature (client: {_get_client_host(request)})')
return JSONResponse(
status_code = 401,
content = {'detail': 'Invalid signature'}
)
return await call_next(request)
@@ -0,0 +1,20 @@
# 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 release, common
__all__ = [
'release',
'common'
]
@@ -0,0 +1,48 @@
# 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 pydantic import BaseModel, HttpUrl
from typing import Optional, List, Literal
__all__ = [
'User',
'Repository',
]
class User(BaseModel):
id: int
login: str
username: Optional[str] = None
full_name: Optional[str] = None
email: Optional[str] = None
class Config:
extra = 'allow'
class Repository(BaseModel):
id: int
name: str
full_name: str
private: bool
html_url: str
clone_url: Optional[str] = None
ssh_url: Optional[str] = None
default_branch: Optional[str] = None
class Config:
extra = 'allow'
@@ -0,0 +1,78 @@
# 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 pydantic import BaseModel, HttpUrl
from typing import Optional, List, Literal
from .common import User, Repository
__all__ = [
'ReleaseAsset',
'Release',
'ReleaseWebhook'
]
class ReleaseAsset(BaseModel):
id: int
name: str
size: Optional[int] = None
download_count: Optional[int] = None
browser_download_url: Optional[HttpUrl] = None
class Config:
extra = 'allow'
class Release(BaseModel):
id: int
tag_name: str
target_commitish: str
name: Optional[str] = None
body: Optional[str] = None
url: str
html_url: str
tarball_url: Optional[HttpUrl] = None
zipball_url: Optional[HttpUrl] = None
upload_url: Optional[HttpUrl] = None
draft: bool
prerelease: bool
created_at: Optional[str] = None
published_at: Optional[str] = None
author: User
assets: List[ReleaseAsset]
class Config:
extra = 'allow'
class ReleaseWebhook(BaseModel):
action: Literal[
'published',
'created',
'edited',
'deleted',
'prereleased',
'released'
]
release: Release
repository: Repository
sender: User
class Config:
extra = 'allow'
@@ -0,0 +1,225 @@
# 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 src.jcloud_deployment_server.integrations.gitea.middlewares.check_enabled import GiteaCheckEnabledMiddleware
from tests.utils.make_request import call_next, make_request
from jcloud_config_parser.ini import INIConfiguration
import pytest
@pytest.mark.asyncio
@pytest.mark.parametrize('path,enabled,should_fail', [
('/gitea', 'false', True),
('/gitea', '', True),
('/gitea', 'f', True),
('/gitea', 'n', True),
('/gitea', '42', True),
('/gitea', 'abcdefg', True),
('/gitea/', 'false', True),
('/gitea/', '', True),
('/gitea/', 'f', True),
('/gitea/', 'n', True),
('/gitea/', '42', True),
('/gitea/', 'abcdefg', True),
('/gitea/42', 'false', True),
('/gitea/42', '', True),
('/gitea/42', 'f', True),
('/gitea/42', 'n', True),
('/gitea/42', '42', True),
('/gitea/42', 'abcdefg', True),
('/gitea/ ', 'false', True),
('/gitea/ ', '', True),
('/gitea/ ', 'f', True),
('/gitea/ ', 'n', True),
('/gitea/ ', '42', True),
('/gitea/ ', 'abcdefg', True),
('/gitea/a', 'false', True),
('/gitea/a', '', True),
('/gitea/a', 'f', True),
('/gitea/a', 'n', True),
('/gitea/a', '42', True),
('/gitea/a', 'abcdefg', True),
('/gitea/a/b', 'false', True),
('/gitea/a/b', '', True),
('/gitea/a/b', 'f', True),
('/gitea/a/b', 'n', True),
('/gitea/a/b', '42', True),
('/gitea/a/b', 'abcdefg', True),
('/gitea/a/', 'false', True),
('/gitea/a/', '', True),
('/gitea/a/', 'f', True),
('/gitea/a/', 'n', True),
('/gitea/a/', '42', True),
('/gitea/a/', 'abcdefg', True),
('/gitea/a/b/', 'false', True),
('/gitea/a/b/', '', True),
('/gitea/a/b/', 'f', True),
('/gitea/a/b/', 'n', True),
('/gitea/a/b/', '42', True),
('/gitea/a/b/', 'abcdefg', True),
('/gite', 'false', False),
('/gite', '', False),
('/gite', 'f', False),
('/gite', 'n', False),
('/gite', '42', False),
('/gite', 'abcdefg', False),
('/g', 'false', False),
('/g', '', False),
('/g', 'f', False),
('/g', 'n', False),
('/g', '42', False),
('/g', 'abcdefg', False),
('/a', 'false', False),
('/a', '', False),
('/a', 'f', False),
('/a', 'n', False),
('/a', '42', False),
('/a', 'abcdefg', False),
('/a/', 'false', False),
('/a/', '', False),
('/a/', 'f', False),
('/a/', 'n', False),
('/a/', '42', False),
('/a/', 'abcdefg', False),
('/a/b', 'false', False),
('/a/b', '', False),
('/a/b', 'f', False),
('/a/b', 'n', False),
('/a/b', '42', False),
('/a/b', 'abcdefg', False),
('/a/b/', 'false', False),
('/a/b/', '', False),
('/a/b/', 'f', False),
('/a/b/', 'n', False),
('/a/b/', '42', False),
('/a/b/', 'abcdefg', False),
('/', 'false', False),
('/', '', False),
('/', 'f', False),
('/', 'n', False),
('/', '42', False),
('/', 'abcdefg', False),
('/gitea', 'true', False),
('/gitea', 'yes', False),
('/gitea', 't', False),
('/gitea', 'y', False),
('/gitea/', 'true', False),
('/gitea/', 'yes', False),
('/gitea/', 't', False),
('/gitea/', 'y', False),
('/gitea/42', 'true', False),
('/gitea/42', 'yes', False),
('/gitea/42', 't', False),
('/gitea/42', 'y', False),
('/gitea/ ', 'true', False),
('/gitea/ ', 'yes', False),
('/gitea/ ', 't', False),
('/gitea/ ', 'y', False),
('/gitea/a', 'true', False),
('/gitea/a', 'yes', False),
('/gitea/a', 't', False),
('/gitea/a', 'y', False),
('/gitea/a/b', 'true', False),
('/gitea/a/b', 'yes', False),
('/gitea/a/b', 't', False),
('/gitea/a/b', 'y', False),
('/gitea/a/', 'true', False),
('/gitea/a/', 'yes', False),
('/gitea/a/', 't', False),
('/gitea/a/', 'y', False),
('/gitea/a/b/', 'true', False),
('/gitea/a/b/', 'yes', False),
('/gitea/a/b/', 't', False),
('/gitea/a/b/', 'y', False),
('/gite', 'true', False),
('/gite', 'yes', False),
('/gite', 't', False),
('/gite', 'y', False),
('/g', 'true', False),
('/g', 'yes', False),
('/g', 't', False),
('/g', 'y', False),
('/a', 'true', False),
('/a', 'yes', False),
('/a', 't', False),
('/a', 'y', False),
('/a/', 'true', False),
('/a/', 'yes', False),
('/a/', 't', False),
('/a/', 'y', False),
('/a/b', 'true', False),
('/a/b', 'yes', False),
('/a/b', 't', False),
('/a/b', 'y', False),
('/a/b/', 'true', False),
('/a/b/', 'yes', False),
('/a/b/', 't', False),
('/a/b/', 'y', False),
('/', 'true', False),
('/', 'yes', False),
('/', 't', False),
('/', 'y', False),
])
async def test_GiteaCheckEnabledMiddleware(path, enabled, should_fail):
configuration = INIConfiguration()
gitea_config = INIConfiguration()
gitea_config.enabled = enabled
configuration.gitea = gitea_config
middleware = GiteaCheckEnabledMiddleware(None, configuration)
req = make_request(
b'',
path
)
res = await middleware.dispatch(req, call_next)
if should_fail:
assert res.status_code == 403
else:
assert hasattr(res, 'called')
@@ -0,0 +1,117 @@
# 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 src.jcloud_deployment_server.integrations.gitea.middlewares.signature import GiteaSignatureMiddleware
from tests.utils.make_request import make_request, call_next
import pytest
import hashlib
import hmac
def make_signature(body: bytes, secret: bytes) -> str:
signature = hmac.new(
secret,
body,
hashlib.sha256
).hexdigest()
return f'sha256={signature}'
@pytest.mark.asyncio
@pytest.mark.parametrize('body,secret', [
(
b'{"event":"push"}',
b'\xa1\xd6h\x0c\xe6\xc0\x99\x82yd\x14\xfew\xcc\x8e\xb0\xf9\x8f\xe6yM\xe5\xdd4\xdc\xb5M+\xef\xc8O\x94'
),
(
b'{"event":"push"}',
b''
),
(
b'',
b'\xa1\xd6h\x0c\xe6\xc0\x99\x82yd\x14\xfew\xcc\x8e\xb0\xf9\x8f\xe6yM\xe5\xdd4\xdc\xb5M+\xef\xc8O\x94'
),
(
b'',
b''
),
(
b'',
b'\x42'
),
(
b'',
b'\x42'
),
(
b'',
b'\x04\x02'
),
(
b'',
b'\x04\x02'
),
])
async def test_GiteaSignatureMiddleware_valid_signature(body, secret):
middleware = GiteaSignatureMiddleware(app = None, secret = secret)
headers = {
'X-Gitea-Signature': make_signature(body, secret)
}
req = make_request(
body,
'/gitea/webhook',
headers
)
res = await middleware.dispatch(req, call_next)
assert hasattr(res, 'called')
@pytest.mark.asyncio
@pytest.mark.parametrize('signature,body,secret', [
(make_signature(b'body', b''), b'body', b'\x42'),
(make_signature(b'body', b'\x42'), b'body', b''),
(make_signature(b'body', b'\x43'), b'body', b'\x42'),
(make_signature(b'body', b'\x42'), b'body', b'\x43'),
(make_signature(b'body', b''), b'body', b'\x42\x43'),
(make_signature(b'body', b'\x42\x43'), b'body', b''),
(make_signature(b'body', b'\x43\x44'), b'body', b'\x42'),
(make_signature(b'body', b'\x42'), b'body', b'\x43\x44'),
(make_signature(b'', b''), b'', b'\x42'),
(make_signature(b'', b'\x42'), b'', b''),
(make_signature(b'', b'\x43'), b'', b'\x42'),
(make_signature(b'', b'\x42'), b'', b'\x43'),
(make_signature(b'', b''), b'', b'\x42\x43'),
(make_signature(b'', b'\x42\x43'), b'', b''),
(make_signature(b'', b'\x43\x44'), b'', b'\x42'),
(make_signature(b'', b'\x42'), b'', b'\x43\x44'),
(make_signature(b'a', b'\x42'), b'', b'\x42'),
(make_signature(b'a', b'\x42'), b'b', b'\x42'),
(make_signature(b'', b'\x42'), b'b', b'\x42'),
])
async def test_GiteaSignatureMiddleware_invalid_signature(signature, body, secret):
middleware = GiteaSignatureMiddleware(app = None, secret = secret)
headers = {
'X-Gitea-Signature': signature
}
req = make_request(
body,
'/gitea/webhook',
headers,
)
res = await middleware.dispatch(req, call_next)
assert res.status_code == 401
+60
View File
@@ -0,0 +1,60 @@
# 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 starlette.requests import Request
__all__ = [
'call_next',
'make_request'
]
class _DummyResponse:
def __init__(self):
self.called = True
async def call_next(request):
return _DummyResponse()
def make_request(body: bytes, path: str, headers: dict = None) -> Request:
'''
Makes a request.
:param body: The request body.
:type body: bytes
:param path: The path.
:type path: str
:param headers: The headers.
:type headers: dict
:return: The request.
:rtype: Request
'''
scope = {
'type': 'http',
'method': 'POST',
'path': path,
'headers': [
(k.lower().encode(), v.encode())
for k, v in (headers or {}).items()
]
}
async def receive():
return {
'type': 'http.request',
'body': body
}
return Request(scope, receive)