Elgato_dark/bunkerweb
1
0
Fork 0
mirror of https://github.com/bunkerity/bunkerweb synced 2026-05-24 09:28:37 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

lint python files

Browse source
This commit is contained in:
Théophile Diot 2023-08-08 11:39:43 +00:00 committed by GitHub
parent c00157ef32
commit 18ee159711
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
25 changed files with 88 additions and 112 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
src/autoconf/IngressController.py
View file

@ -31,7 +31,9 @@ class IngressController(Controller):
def _to_instances(self, controller_instance) -> List[dict]:
instance = {}
instance["name"] = controller_instance.metadata.name
instance["hostname"] = controller_instance.status.pod_ip or controller_instance.metadata.name
instance["hostname"] = (
controller_instance.status.pod_ip or controller_instance.metadata.name
)
health = False
if controller_instance.status.conditions:
for condition in controller_instance.status.conditions:

6
src/common/core/blacklist/jobs/blacklist-download.py
View file

@ -77,11 +77,7 @@ try:
logger.info("Blacklist is not activated, skipping downloads...")
_exit(0)
db = Database(
logger,
sqlalchemy_string=getenv("DATABASE_URI", None),
pool=False
)
db = Database(logger, sqlalchemy_string=getenv("DATABASE_URI", None), pool=False)
# Create directories if they don't exist
blacklist_path = Path(sep, "var", "cache", "bunkerweb", "blacklist")

6
src/common/core/bunkernet/jobs/bunkernet-data.py
View file

@ -57,11 +57,7 @@ try:
# Get ID from cache
bunkernet_id = None
db = Database(
logger,
sqlalchemy_string=getenv("DATABASE_URI", None),
pool=False
)
db = Database(logger, sqlalchemy_string=getenv("DATABASE_URI", None), pool=False)
bunkernet_id = get_file_in_db("instance.id", db)
if bunkernet_id:
bunkernet_path.joinpath("bunkernet.id").write_bytes(bunkernet_id)

6
src/common/core/bunkernet/jobs/bunkernet-register.py
View file

@ -58,11 +58,7 @@ try:
# Get ID from cache
bunkernet_id = None
db = Database(
logger,
sqlalchemy_string=getenv("DATABASE_URI", None),
pool=False
)
db = Database(logger, sqlalchemy_string=getenv("DATABASE_URI", None), pool=False)
bunkernet_id = get_file_in_db("instance.id", db)
if bunkernet_id:
bunkernet_path.joinpath("bunkernet.id").write_bytes(bunkernet_id)

8
src/common/core/customcert/jobs/custom-cert.py
View file

@ -109,9 +109,7 @@ try:
if getenv("USE_CUSTOM_SSL", "no") == "yes" and getenv("SERVER_NAME", "") != "":
db = Database(
logger,
sqlalchemy_string=getenv("DATABASE_URI", None),
pool=False
logger, sqlalchemy_string=getenv("DATABASE_URI", None), pool=False
)
cert_path = getenv("CUSTOM_SSL_CERT", "")
@ -141,9 +139,7 @@ try:
if not db:
db = Database(
logger,
sqlalchemy_string=getenv("DATABASE_URI", None),
pool=False
logger, sqlalchemy_string=getenv("DATABASE_URI", None), pool=False
)
cert_path = getenv(f"{first_server}_CUSTOM_SSL_CERT", "")

6
src/common/core/greylist/jobs/greylist-download.py
View file

@ -77,11 +77,7 @@ try:
logger.info("Greylist is not activated, skipping downloads...")
_exit(0)
db = Database(
logger,
sqlalchemy_string=getenv("DATABASE_URI", None),
pool=False
)
db = Database(logger, sqlalchemy_string=getenv("DATABASE_URI", None), pool=False)
# Create directories if they don't exist
greylist_path = Path(sep, "var", "cache", "bunkerweb", "greylist")

6
src/common/core/jobs/jobs/download-plugins.py
View file

@ -174,11 +174,7 @@ try:
external_plugins.append(plugin_file)
external_plugins_ids.append(plugin_file["id"])
db = Database(
logger,
sqlalchemy_string=getenv("DATABASE_URI"),
pool=False
)
db = Database(logger, sqlalchemy_string=getenv("DATABASE_URI"), pool=False)
lock = Lock()
for plugin in db.get_plugins(external=True, with_data=True):

6
src/common/core/jobs/jobs/mmdb-asn.py
View file

@ -68,11 +68,7 @@ try:
"Unable to check if asn.mmdb is the latest version, downloading it anyway..."
)
db = Database(
logger,
sqlalchemy_string=getenv("DATABASE_URI", None),
pool=False
)
db = Database(logger, sqlalchemy_string=getenv("DATABASE_URI", None), pool=False)
if dl_mmdb:
# Don't go further if the cache is fresh

6
src/common/core/jobs/jobs/mmdb-country.py
View file

@ -68,11 +68,7 @@ try:
"Unable to check if country.mmdb is the latest version, downloading it anyway..."
)
db = Database(
logger,
sqlalchemy_string=getenv("DATABASE_URI", None),
pool=False
)
db = Database(logger, sqlalchemy_string=getenv("DATABASE_URI", None), pool=False)
if dl_mmdb:
# Don't go further if the cache is fresh

4
src/common/core/letsencrypt/jobs/certbot-auth.py
View file

@ -50,9 +50,7 @@ try:
# Cluster case
if bw_integration in ("Docker", "Swarm", "Kubernetes", "Autoconf"):
db = Database(
logger,
sqlalchemy_string=getenv("DATABASE_URI", None),
pool=False
logger, sqlalchemy_string=getenv("DATABASE_URI", None), pool=False
)
lock = Lock()

4
src/common/core/letsencrypt/jobs/certbot-cleanup.py
View file

@ -49,9 +49,7 @@ try:
# Cluster case
if bw_integration in ("Docker", "Swarm", "Kubernetes", "Autoconf"):
db = Database(
logger,
sqlalchemy_string=getenv("DATABASE_URI", None),
pool=False
logger, sqlalchemy_string=getenv("DATABASE_URI", None), pool=False
)
lock = Lock()
with lock:

4
src/common/core/letsencrypt/jobs/certbot-deploy.py
View file

@ -65,9 +65,7 @@ try:
files = {"archive.tar.gz": tgz}
db = Database(
logger,
sqlalchemy_string=getenv("DATABASE_URI", None),
pool=False
logger, sqlalchemy_string=getenv("DATABASE_URI", None), pool=False
)
lock = Lock()

29
src/common/core/letsencrypt/jobs/certbot-new.py
View file

@ -55,7 +55,7 @@ def certbot_new(
"--email",
email,
"--agree-tos",
"--expand"
"--expand",
]
+ (["--staging"] if getenv("USE_LETS_ENCRYPT_STAGING", "no") == "yes" else []),
stdin=DEVNULL,
@ -64,6 +64,7 @@ def certbot_new(
| {"PYTHONPATH": join(sep, "usr", "share", "bunkerweb", "deps", "python")},
).returncode
def certbot_check_domains(domains: list[str], letsencrypt_path: Path) -> int:
proc = run(
[
@ -90,10 +91,14 @@ def certbot_check_domains(domains: list[str], letsencrypt_path: Path) -> int:
needed_domains = set(domains)
for raw_domains in findall(r"^ Domains: (.*)$", proc.stdout, MULTILINE):
current_domains = raw_domains.split(" ")
if current_domains[0] == first_needed_domain and set(current_domains) == needed_domains:
if (
current_domains[0] == first_needed_domain
and set(current_domains) == needed_domains
):
return 1
return 0
status = 0
try:
@ -143,11 +148,7 @@ try:
bw_integration = "Docker"
# Extract letsencrypt folder if it exists in db
db = Database(
logger,
sqlalchemy_string=getenv("DATABASE_URI", None),
pool=False
)
db = Database(logger, sqlalchemy_string=getenv("DATABASE_URI", None), pool=False)
tgz = get_file_in_db("folder.tgz", db, job_name="certbot-renew")
if tgz:
@ -193,7 +194,12 @@ try:
f"Asking certificates for domains : {domains} (email = {real_email}) ...",
)
if (
certbot_new(domains.replace(" ", ","), real_email, letsencrypt_path, letsencrypt_job_path)
certbot_new(
domains.replace(" ", ","),
real_email,
letsencrypt_path,
letsencrypt_job_path,
)
!= 0
):
status = 2
@ -225,7 +231,12 @@ try:
f"Asking certificates for domain(s) : {domains} (email = {real_email}) ...",
)
if (
certbot_new(domains.replace(" ", ","), real_email, letsencrypt_path, letsencrypt_job_path)
certbot_new(
domains.replace(" ", ","),
real_email,
letsencrypt_path,
letsencrypt_job_path,
)
!= 0
):
status = 2

6
src/common/core/letsencrypt/jobs/certbot-renew.py
View file

@ -105,11 +105,7 @@ try:
bw_integration = "Docker"
# Extract letsencrypt folder if it exists in db
db = Database(
logger,
sqlalchemy_string=getenv("DATABASE_URI", None),
pool=False
)
db = Database(logger, sqlalchemy_string=getenv("DATABASE_URI", None), pool=False)
tgz = get_file_in_db("folder.tgz", db)
if tgz:

4
src/common/core/misc/jobs/default-server-cert.py
View file

@ -100,9 +100,7 @@ try:
)
db = Database(
logger,
sqlalchemy_string=getenv("DATABASE_URI", None),
pool=False
logger, sqlalchemy_string=getenv("DATABASE_URI", None), pool=False
)
cached, err = set_file_in_db(

6
src/common/core/realip/jobs/realip-download.py
View file

@ -73,11 +73,7 @@ try:
tmp_realip_path = Path(sep, "var", "tmp", "bunkerweb", "realip")
tmp_realip_path.mkdir(parents=True, exist_ok=True)
db = Database(
logger,
sqlalchemy_string=getenv("DATABASE_URI", None),
pool=False
)
db = Database(logger, sqlalchemy_string=getenv("DATABASE_URI", None), pool=False)
# Don't go further if the cache is fresh
if is_cached_file(realip_path.joinpath("combined.list"), "hour", db):

8
src/common/core/selfsigned/jobs/self-signed.py
View file

@ -132,9 +132,7 @@ try:
if not db:
db = Database(
logger,
sqlalchemy_string=getenv("DATABASE_URI", None),
pool=False
logger, sqlalchemy_string=getenv("DATABASE_URI", None), pool=False
)
ret, ret_status = generate_cert(
@ -154,9 +152,7 @@ try:
# Singlesite case
elif getenv("GENERATE_SELF_SIGNED_SSL", "no") == "yes" and getenv("SERVER_NAME"):
db = Database(
logger,
sqlalchemy_string=getenv("DATABASE_URI", None),
pool=False
logger, sqlalchemy_string=getenv("DATABASE_URI", None), pool=False
)
first_server = getenv("SERVER_NAME", "").split(" ")[0]

6
src/common/core/whitelist/jobs/whitelist-download.py
View file

@ -77,11 +77,7 @@ try:
logger.info("Whitelist is not activated, skipping downloads...")
_exit(0)
db = Database(
logger,
sqlalchemy_string=getenv("DATABASE_URI", None),
pool=False
)
db = Database(logger, sqlalchemy_string=getenv("DATABASE_URI", None), pool=False)
# Create directories if they don't exist
whitelist_path = Path(sep, "var", "cache", "bunkerweb", "whitelist")

24
src/common/db/Database.py
View file

@ -62,7 +62,7 @@ class Database:
sqlalchemy_string: Optional[str] = None,
*,
ui: bool = False,
pool: bool = True
pool: bool = True,
) -> None:
"""Initialize the database"""
self.__logger = logger
@ -95,7 +95,7 @@ class Database:
sqlalchemy_string,
future=True,
poolclass=None if pool else NullPool,
pool_pre_ping=True
pool_pre_ping=True,
)
except ArgumentError:
self.__logger.error(f"Invalid database URI: {sqlalchemy_string}")
@ -140,7 +140,7 @@ class Database:
sqlalchemy_string,
future=True,
poolclass=None if pool else NullPool,
pool_pre_ping=True
pool_pre_ping=True,
)
if "Unknown table" in str(e):
not_connected = False
@ -311,7 +311,7 @@ class Database:
Metadata.custom_configs_changed,
Metadata.external_plugins_changed,
Metadata.config_changed,
Metadata.instances_changed
Metadata.instances_changed,
)
.filter_by(id=1)
.first()
@ -323,14 +323,20 @@ class Database:
external_plugins_changed=metadata is not None
and metadata.external_plugins_changed,
config_changed=metadata is not None and metadata.config_changed,
instances_changed=metadata is not None and metadata.instances_changed
instances_changed=metadata is not None
and metadata.instances_changed,
)
except BaseException:
return format_exc()
def checked_changes(self, changes: Optional[List[str]] = None) -> str:
"""Set that the config, the custom configs, the plugins and instances didn't change"""
changes = changes or ["config", "custom_configs", "external_plugins", "instances"]
changes = changes or [
"config",
"custom_configs",
"external_plugins",
"instances",
]
with self.__db_session() as session:
try:
metadata = session.query(Metadata).get(1)
@ -659,8 +665,6 @@ class Database:
)
)
#config.pop("SERVER_NAME", None)
for key, value in config.items():
suffix = 0
if self.suffix_rx.search(key):
@ -922,7 +926,9 @@ class Database:
)
if config["MULTISITE"] == "yes":
servers = " ".join(service.id for service in session.query(Services).all())
servers = " ".join(
service.id for service in session.query(Services).all()
)
config["SERVER_NAME"] = (
servers
if methods is False

10
src/common/gen/save_config.py
View file

@ -61,11 +61,7 @@ def get_instance_configs_and_apis(instance: Any, db, _type="Docker"):
tmp_config[splitted[0]] = splitted[1]
if not db and splitted[0] == "DATABASE_URI":
db = Database(
logger,
sqlalchemy_string=splitted[1],
pool=False
)
db = Database(logger, sqlalchemy_string=splitted[1], pool=False)
elif splitted[0] == "API_HTTP_PORT":
api_http_port = splitted[1]
elif splitted[0] == "API_SERVER_NAME":
@ -268,9 +264,7 @@ if __name__ == "__main__":
if not db and splitted[0] == "DATABASE_URI":
db = Database(
logger,
sqlalchemy_string=splitted[1],
pool=False
logger, sqlalchemy_string=splitted[1], pool=False
)
elif splitted[0] == "API_HTTP_PORT":
api_http_port = splitted[1]

6
src/scheduler/JobScheduler.py
View file

@ -32,6 +32,7 @@ from logger import setup_logger # type: ignore
from ApiCaller import ApiCaller # type: ignore
from API import API # type: ignore
class JobScheduler(ApiCaller):
def __init__(
self,
@ -77,7 +78,10 @@ class JobScheduler(ApiCaller):
with self.__thread_lock:
instances = self.__db.get_instances()
for instance in instances:
api = API(f"http://{instance['hostname']}:{instance['port']}", host=instance["server_name"])
api = API(
f"http://{instance['hostname']}:{instance['port']}",
host=instance["server_name"],
)
apis.append(api)
except:
self.__logger.warning(

15
src/scheduler/main.py
View file

@ -175,16 +175,20 @@ def dict_to_frozenset(d):
return frozenset((k, dict_to_frozenset(v)) for k, v in d.items())
return d
def api_to_instance(api):
hostname_port = api.endpoint.replace("http://", "").replace("https://", "").replace("/", "").split(":")
hostname_port = (
api.endpoint.replace("http://", "")
.replace("https://", "")
.replace("/", "")
.split(":")
)
return {
"hostname": hostname_port[0],
"env": {
"API_HTTP_PORT": int(hostname_port[1]),
"API_SERVER_NAME": api.host
}
"env": {"API_HTTP_PORT": int(hostname_port[1]), "API_SERVER_NAME": api.host},
}
def update_docker_instances(sc, db, force=False):
current_apis = set(sc.apis)
sc.auto_setup()
@ -195,6 +199,7 @@ def update_docker_instances(sc, db, force=False):
return db.update_instances(new_instances)
return ""
if __name__ == "__main__":
try:
# Don't execute if pid file exists

2
src/ui/gunicorn.conf.py
View file

@ -15,4 +15,4 @@ worker_class = "gevent"
threads = 1
workers = 1
graceful_timeout = 0
secure_scheme_headers = {}
secure_scheme_headers = {}

15
src/ui/main.py
View file

@ -85,6 +85,7 @@ from logger import setup_logger # type: ignore
from Database import Database # type: ignore
from logging import getLogger
def stop_gunicorn():
p = Popen(["pgrep", "-f", "gunicorn"], stdout=PIPE)
out, _ = p.communicate()
@ -119,8 +120,14 @@ app = Flask(
template_folder="templates",
)
PROXY_NUMBERS = int(getenv("PROXY_NUMBERS", "1"))
app.wsgi_app = ReverseProxied(app.wsgi_app, x_for=PROXY_NUMBERS, x_proto=PROXY_NUMBERS, x_host=PROXY_NUMBERS, x_prefix=PROXY_NUMBERS)
gunicorn_logger = getLogger('gunicorn.error')
app.wsgi_app = ReverseProxied(
app.wsgi_app,
x_for=PROXY_NUMBERS,
x_proto=PROXY_NUMBERS,
x_host=PROXY_NUMBERS,
x_prefix=PROXY_NUMBERS,
)
gunicorn_logger = getLogger("gunicorn.error")
app.logger.handlers = gunicorn_logger.handlers
app.logger.setLevel(gunicorn_logger.level)
@ -1526,7 +1533,9 @@ def darkmode():
def check_reloading():
if not app.config["RELOADING"] or app.config["LAST_RELOAD"] + 60 < time():
if app.config["RELOADING"]:
app.logger.warning("Reloading took too long, forcing the state to be reloaded")
app.logger.warning(
"Reloading took too long, forcing the state to be reloaded"
)
flash("Forced the status to be reloaded", "error")
app.config["RELOADING"] = False

3
tests/LinuxTest.py
View file

@ -112,7 +112,8 @@ class LinuxTest(Test):
setup = f"{test}/setup-linux.sh"
if isfile(setup):
proc = self.docker_exec(
self.__distro, f"cd /opt/{self._name} && ./setup-linux.sh && chown -R nginx:nginx /etc/bunkerweb/configs"
self.__distro,
f"cd /opt/{self._name} && ./setup-linux.sh && chown -R nginx:nginx /etc/bunkerweb/configs",
)
if proc.returncode != 0:
raise Exception("docker exec setup failed (test)")