First working version

This commit is contained in:
2026-05-21 05:02:40 +00:00
parent 60cdacb9d6
commit f204927482
52 changed files with 3046 additions and 12 deletions
+91
View File
@@ -0,0 +1,91 @@
## Generic Parts
# These functions are providing the functionality to load
# arbitrary configuration files.
#
# They can be imported by other code (see `ldap_config.py` for an example).
import importlib.util
import sys
from os import scandir
from os.path import abspath, isfile
def _filename(f):
return f.name
def _import(module_name, path, loaded_configurations):
spec = importlib.util.spec_from_file_location('', path)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
sys.modules[module_name] = module
loaded_configurations.insert(0, module)
print(f"🧬 loaded config '{path}'")
def read_configurations(config_module, config_dir, main_config):
loaded_configurations = []
main_config_path = abspath(f'{config_dir}/{main_config}.py')
if isfile(main_config_path):
_import(f'{config_module}.{main_config}', main_config_path, loaded_configurations)
else:
print(f"⚠️ Main configuration '{main_config_path}' not found.")
with scandir(config_dir) as it:
for f in sorted(it, key=_filename):
if not f.is_file():
continue
if f.name.startswith('__'):
continue
if not f.name.endswith('.py'):
continue
if f.name == f'{main_config}.py':
continue
if f.name == f'{config_dir}.py':
continue
module_name = f'{config_module}.{f.name[: -len(".py")]}'.replace('.', '_')
_import(module_name, f.path, loaded_configurations)
if len(loaded_configurations) == 0:
print(f"‼️ No configuration files found in '{config_dir}'.")
raise ImportError(f"No configuration files found in '{config_dir}'.")
return loaded_configurations
## Specific Parts
# This section's code actually loads the various configuration files
# into the module with the given name.
# It contains the logic to resolve arbitrary configuration options by
# levaraging dynamic programming using `__getattr__`.
_loaded_configurations = read_configurations(
config_dir='/etc/netbox/config/',
config_module='netbox.configuration',
main_config='configuration',
)
def __getattr__(name):
for config in _loaded_configurations:
try:
return getattr(config, name)
except:
pass
raise AttributeError
def __dir__():
names = []
for config in _loaded_configurations:
names.extend(config.__dir__())
return names
+65
View File
@@ -0,0 +1,65 @@
#!/bin/bash
# Runs on every start of the NetBox Docker container
# Stop when an error occures
set -e
# Allows NetBox to be run as non-root users
umask 002
# Load correct Python3 env
# shellcheck disable=SC1091
source /opt/netbox/venv/bin/activate
# Try to connect to the DB
DB_WAIT_TIMEOUT=${DB_WAIT_TIMEOUT-3}
MAX_DB_WAIT_TIME=${MAX_DB_WAIT_TIME-30}
CUR_DB_WAIT_TIME=0
while [ "${CUR_DB_WAIT_TIME}" -lt "${MAX_DB_WAIT_TIME}" ]; do
# Read and truncate connection error tracebacks to last line by default
exec {psfd}< <(./manage.py showmigrations 2>&1)
read -rd '' DB_ERR <&$psfd || :
exec {psfd}<&-
wait $! && break
if [ -n "$DB_WAIT_DEBUG" ]; then
echo "$DB_ERR"
else
readarray -tn 0 DB_ERR_LINES <<<"$DB_ERR"
echo "${DB_ERR_LINES[@]: -1}"
echo "[ Use DB_WAIT_DEBUG=1 in netbox.env to print full traceback for errors here ]"
fi
echo "⏳ Waiting on DB... (${CUR_DB_WAIT_TIME}s / ${MAX_DB_WAIT_TIME}s)"
sleep "${DB_WAIT_TIMEOUT}"
CUR_DB_WAIT_TIME=$((CUR_DB_WAIT_TIME + DB_WAIT_TIMEOUT))
done
if [ "${CUR_DB_WAIT_TIME}" -ge "${MAX_DB_WAIT_TIME}" ]; then
echo "❌ Waited ${MAX_DB_WAIT_TIME}s or more for the DB to become ready."
exit 1
fi
# Check if update is needed
if ! ./manage.py migrate --check >/dev/null 2>&1; then
echo "⚙️ Applying database migrations"
./manage.py migrate --no-input
echo "⚙️ Running trace_paths"
./manage.py trace_paths --no-input
echo "⚙️ Removing stale content types"
./manage.py remove_stale_contenttypes --no-input
echo "⚙️ Removing expired user sessions"
./manage.py clearsessions
echo "⚙️ Building search index (lazy)"
./manage.py reindex --lazy
fi
# Create Superuser if required
if [ "$SKIP_SUPERUSER" == "true" ]; then
echo "↩️ Skip creating the superuser"
else
./manage.py shell --no-startup --no-imports --interface python \
</opt/netbox/super_user.py
fi
echo "✅ Initialisation is done."
# Launch whatever is passed by docker
# (i.e. the RUN instruction in the Dockerfile)
exec "$@"
+13
View File
@@ -0,0 +1,13 @@
from granian.utils.proxies import wrap_wsgi_with_proxy_headers
from netbox.wsgi import application
application = wrap_wsgi_with_proxy_headers(
application,
trusted_hosts=[
'10.0.0.0/8',
'172.16.0.0/12',
'192.168.0.0/16',
'fc00::/7',
'fe80::/10',
],
)
+21
View File
@@ -0,0 +1,21 @@
#!/bin/bash
exec granian \
--host "::" \
--port "8080" \
--interface "wsgi" \
--no-ws \
--workers "${GRANIAN_WORKERS:-4}" \
--respawn-failed-workers \
--backpressure "${GRANIAN_BACKPRESSURE:-${GRANIAN_WORKERS:-4}}" \
--loop "uvloop" \
--log \
--log-level "info" \
--access-log \
--working-dir "/opt/netbox/netbox/" \
--static-path-route "/static" \
--static-path-mount "/opt/netbox/netbox/static/" \
--static-path-dir-to-file index.html \
--pid-file "/tmp/granian.pid" \
"${GRANIAN_EXTRA_ARGS[@]}" \
"netbox.granian:application"
+23
View File
@@ -0,0 +1,23 @@
from .configuration import read_configurations
_loaded_configurations = read_configurations(
config_dir='/etc/netbox/config/ldap/',
config_module='netbox.configuration.ldap',
main_config='ldap_config',
)
def __getattr__(name):
for config in _loaded_configurations:
try:
return getattr(config, name)
except:
pass
raise AttributeError
def __dir__():
names = []
for config in _loaded_configurations:
names.extend(config.__dir__())
return names
+45
View File
@@ -0,0 +1,45 @@
import sys
from os import environ
from django.conf import settings
from users.choices import TokenVersionChoices
from users.models import Token, User
# Read secret from file
def _read_secret(secret_name: str, default: str | None = None) -> str | None:
try:
with open('/run/secrets/' + secret_name, encoding='utf-8') as f:
return f.readline().strip()
except OSError:
return default
su_name = environ.get('SUPERUSER_NAME', 'admin')
su_email = environ.get('SUPERUSER_EMAIL', 'admin@example.com')
su_password = _read_secret('superuser_password', environ.get('SUPERUSER_PASSWORD'))
su_api_token = _read_secret('superuser_api_token', environ.get('SUPERUSER_API_TOKEN'))
su_api_key = _read_secret('superuser_api_key', environ.get('SUPERUSER_API_KEY'))
if User.objects.filter(username=su_name).exists():
print(f'User with name "{su_name}" already exists.')
sys.exit(0)
if not su_password:
print(
'⚠️ Warning: No superuser password provided. Please set the SUPERUSER_PASSWORD environment variable or provide a secret file. The superuser will not be created.'
)
sys.exit(0)
u = User.objects.create_superuser(su_name, su_email, su_password)
if not settings.API_TOKEN_PEPPERS:
print('⚠️ No API token was created as API_TOKEN_PEPPERS is not set')
print(f'💡 Superuser Username: {su_name}, E-Mail: {su_email}')
else:
if su_api_key and su_api_token:
t = Token.objects.create(user=u, token=su_api_token, version=TokenVersionChoices.V2, key=su_api_key)
print(f'💡 Superuser Username: {su_name}, E-Mail: {su_email},')
print(f"💡 API Token: use with '{t.get_auth_header_prefix()}<Your token>'")
else:
print('⚠️ No API token was created for the superuser as SUPERUSER_API_TOKEN and SUPERUSER_API_KEY are not set')
print(f'💡 Superuser Username: {su_name}, E-Mail: {su_email}')