generated from jCloud/repository-template
Compare commits
30 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
7c6ca3b5a4
|
|||
|
b8dc240a8a
|
|||
|
547edfaef0
|
|||
|
7ca1d4c609
|
|||
|
791e1d972b
|
|||
|
0e23b4d349
|
|||
|
c26291bb1c
|
|||
|
97f1a802fa
|
|||
|
16bd8b25f2
|
|||
|
944c41095d
|
|||
|
6afa9af3b2
|
|||
|
c9e317f7ca
|
|||
|
2da8da9f26
|
|||
|
2db98caaef
|
|||
|
4cdd52059d
|
|||
|
b295738c0e
|
|||
|
c0cec78b03
|
|||
|
81f79783b6
|
|||
|
d4e7d983fa
|
|||
|
044ccba66a
|
|||
|
10a25b82af
|
|||
|
374dc89c09
|
|||
|
157d054854
|
|||
|
58700f2390
|
|||
|
23faf74092
|
|||
|
30a982c95d
|
|||
|
d9fa676605
|
|||
|
3d0dc70db5
|
|||
|
6ed50eb102
|
|||
|
1b26f5c86f
|
@@ -1,4 +1,4 @@
|
||||
Copyright 2026 jCloud Services GbR
|
||||
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.
|
||||
|
||||
+23
-2
@@ -23,8 +23,29 @@
|
||||
# openapi=
|
||||
|
||||
[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=
|
||||
|
||||
# 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=
|
||||
@@ -11,5 +11,11 @@ redoc=/redoc
|
||||
openapi=/openapi.json
|
||||
|
||||
[logging]
|
||||
logfile=/proc/self/fd/1
|
||||
loglevel=INFO
|
||||
logtarget=stdout
|
||||
logfile=
|
||||
fail_strict=false
|
||||
loglevel=INFO
|
||||
|
||||
[gitea]
|
||||
enabled=false
|
||||
webhook_secret_file=
|
||||
+3
-1
@@ -5,4 +5,6 @@
|
||||
### Added
|
||||
|
||||
- Add basic API structure
|
||||
- Add configuration file template
|
||||
- Add configuration file template
|
||||
- Add models for Gitea webhooks
|
||||
- Add middleware for Gitea webhook signatures
|
||||
@@ -1,4 +1,4 @@
|
||||
# Copyright 2026 jCloud Services GbR
|
||||
# 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.
|
||||
@@ -12,144 +12,7 @@
|
||||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
|
||||
from fastapi import FastAPI
|
||||
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
|
||||
)
|
||||
from .api import main
|
||||
|
||||
if __name__ == '__main__':
|
||||
uvicorn.run(
|
||||
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
|
||||
},
|
||||
},
|
||||
}
|
||||
)
|
||||
main.main()
|
||||
@@ -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)
|
||||
@@ -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
|
||||
@@ -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
|
||||
)
|
||||
@@ -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'
|
||||
+225
@@ -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
|
||||
@@ -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)
|
||||
Reference in New Issue
Block a user