Pleasing the linter

This commit is contained in:
theo@manjaro 2022-07-15 15:11:45 +02:00
parent eb3c391624
commit 5f2785c6c9
6 changed files with 60 additions and 51 deletions

View file

@ -47,7 +47,6 @@ from yunohost.utils.error import YunohostValidationError, YunohostError
from yunohost.utils.network import get_public_ip
from yunohost.log import is_unit_operation
from yunohost.hook import hook_callback
from yunohost.dyndns import dyndns_update
logger = getActionLogger("yunohost.domain")
@ -625,18 +624,19 @@ def _get_registar_settings(domain):
@is_unit_operation()
def domain_dns_push(operation_logger, domains, dry_run=False, force=False, purge=False, auto=False):
if auto:
domains = domain_list(exclude_subdomains=True,auto_push=True)["domains"]
elif len(domains)==0:
domains = domain_list(exclude_subdomains=True)["domains"]
domains = domain_list(exclude_subdomains=True, auto_push=True)["domains"]
elif len(domains) == 0:
domains = domain_list(exclude_subdomains=True)["domains"]
error_domains = []
for domain in domains:
try:
domain_dns_push_unique(domain,dry_run=dry_run,force=force,purge=purge)
domain_dns_push_unique(domain, dry_run=dry_run, force=force, purge=purge)
except YunohostError as e:
logger.error(m18n.n("domain_dns_push_failed_domain",domain=domain,error=str(e)))
logger.error(m18n.n("domain_dns_push_failed_domain", domain=domain, error=str(e)))
error_domains.append(domain)
if len(error_domains)>0:
raise YunohostError("domain_dns_push_failed_domains",domains=', '.join(error_domains))
if len(error_domains) > 0:
raise YunohostError("domain_dns_push_failed_domains", domains=', '.join(error_domains))
@is_unit_operation()
def domain_dns_push_unique(operation_logger, domain, dry_run=False, force=False, purge=False):
@ -660,9 +660,8 @@ def domain_dns_push_unique(operation_logger, domain, dry_run=False, force=False,
# FIXME: in the future, properly unify this with yunohost dyndns update
if registrar == "yunohost":
#logger.info(m18n.n("domain_dns_registrar_yunohost"))
from yunohost.dyndns import dyndns_update
dyndns_update(domain=domain,force=force)
dyndns_update(domain=domain, force=force)
return {}
if registrar == "parent_domain":

View file

@ -52,7 +52,7 @@ DOMAIN_SETTINGS_DIR = "/etc/yunohost/domains"
domain_list_cache: Dict[str, Any] = {}
def domain_list(exclude_subdomains=False,auto_push=False,full=False):
def domain_list(exclude_subdomains=False, auto_push=False, full=False):
"""
List domains
@ -96,11 +96,11 @@ def domain_list(exclude_subdomains=False,auto_push=False,full=False):
if full:
for i in range(len(result_list)):
domain = result_list[i]
dyndns = is_yunohost_dyndns_domain(domain) and len(domain.split("."))==3
result_list[i] = {'name':domain,'isdyndns': dyndns}
dyndns = is_yunohost_dyndns_domain(domain) and len(domain.split(".")) == 3
result_list[i] = {'name': domain, 'isdyndns': dyndns}
result = {"domains": result_list, "main": _get_maindomain()}
# Cache answer only if not using exclude_subdomains or full
if not (full or exclude_subdomains):
domain_list_cache = result
@ -156,7 +156,7 @@ def domain_add(operation_logger, domain, subscribe=None, no_subscribe=False):
from yunohost.utils.ldap import _get_ldap_interface
from yunohost.certificate import _certificate_install_selfsigned
if subscribe!=0 and subscribe!=None:
if subscribe != 0 and subscribe is not None:
operation_logger.data_to_redact.append(subscribe)
if domain.startswith("xmpp-upload."):
@ -177,9 +177,9 @@ def domain_add(operation_logger, domain, subscribe=None, no_subscribe=False):
domain = domain.encode("idna").decode("utf-8")
# Detect if this is a DynDNS domain ( and not a subdomain of a DynDNS domain )
dyndns = is_yunohost_dyndns_domain(domain) and len(domain.split("."))==3
dyndns = is_yunohost_dyndns_domain(domain) and len(domain.split(".")) == 3
if dyndns:
if ((subscribe==None) == (no_subscribe==False)):
if ((subscribe is None) == (no_subscribe is False)):
raise YunohostValidationError("domain_dyndns_instruction_unclear")
from yunohost.dyndns import is_subscribing_allowed
@ -194,7 +194,7 @@ def domain_add(operation_logger, domain, subscribe=None, no_subscribe=False):
if dyndns and not no_subscribe:
# Actually subscribe
domain_dyndns_subscribe(domain=domain,password=subscribe)
domain_dyndns_subscribe(domain=domain, password=subscribe)
_certificate_install_selfsigned([domain], True)
@ -244,7 +244,7 @@ def domain_add(operation_logger, domain, subscribe=None, no_subscribe=False):
@is_unit_operation()
def domain_remove(operation_logger, domain, remove_apps=False, force=False, unsubscribe=None,no_unsubscribe=False):
def domain_remove(operation_logger, domain, remove_apps=False, force=False, unsubscribe=None, no_unsubscribe=False):
"""
Delete domains
@ -259,8 +259,8 @@ def domain_remove(operation_logger, domain, remove_apps=False, force=False, unsu
from yunohost.hook import hook_callback
from yunohost.app import app_ssowatconf, app_info, app_remove
from yunohost.utils.ldap import _get_ldap_interface
if unsubscribe!=0 and unsubscribe!=None:
if unsubscribe != 0 and unsubscribe is not None:
operation_logger.data_to_redact.append(unsubscribe)
# the 'force' here is related to the exception happening in domain_add ...
@ -322,16 +322,16 @@ def domain_remove(operation_logger, domain, remove_apps=False, force=False, unsu
"domain_uninstall_app_first",
apps="\n".join([x[1] for x in apps_on_that_domain]),
)
# Detect if this is a DynDNS domain ( and not a subdomain of a DynDNS domain )
dyndns = is_yunohost_dyndns_domain(domain) and len(domain.split("."))==3
dyndns = is_yunohost_dyndns_domain(domain) and len(domain.split(".")) == 3
if dyndns:
if ((unsubscribe==None) == (no_unsubscribe==False)):
if ((unsubscribe is None) == (no_unsubscribe is False)):
raise YunohostValidationError("domain_dyndns_instruction_unclear_unsubscribe")
operation_logger.start()
if not dyndns and (unsubscribe!=None or no_unsubscribe!=False):
if not dyndns and ((unsubscribe is not None) or (no_unsubscribe is not False)):
logger.warning("This domain is not a DynDNS one, no need for the --unsubscribe or --no-unsubscribe option")
ldap = _get_ldap_interface()
@ -381,7 +381,7 @@ def domain_remove(operation_logger, domain, remove_apps=False, force=False, unsu
# If a password is provided, delete the DynDNS record
if dyndns and not no_unsubscribe:
# Actually unsubscribe
domain_dyndns_unsubscribe(domain=domain,password=unsubscribe)
domain_dyndns_unsubscribe(domain=domain, password=unsubscribe)
logger.success(m18n.n("domain_deleted"))
@ -394,6 +394,7 @@ def domain_dyndns_subscribe(**kwargs):
dyndns_subscribe(**kwargs)
def domain_dyndns_unsubscribe(**kwargs):
"""
Unsubscribe from a DynDNS domain
@ -402,6 +403,7 @@ def domain_dyndns_unsubscribe(**kwargs):
dyndns_unsubscribe(**kwargs)
def domain_dyndns_list():
"""
Returns all currently subscribed DynDNS domains
@ -410,6 +412,7 @@ def domain_dyndns_list():
return dyndns_list()
def domain_dyndns_update(**kwargs):
"""
Update a DynDNS domain
@ -471,6 +474,7 @@ def domain_url_available(domain, path):
return len(_get_conflicting_apps(domain, path)) == 0
def _get_maindomain():
with open("/etc/yunohost/current_host", "r") as f:
maindomain = f.readline().rstrip()

View file

@ -24,7 +24,6 @@
Subscribe and Update DynDNS Hosts
"""
import os
import re
import json
import glob
import base64
@ -50,6 +49,7 @@ DYNDNS_PROVIDER = "dyndns.yunohost.org"
DYNDNS_DNS_AUTH = ["ns0.yunohost.org", "ns1.yunohost.org"]
MAX_DYNDNS_DOMAINS = 1
def is_subscribing_allowed():
"""
Check if the limit of subscribed DynDNS domains has been reached
@ -57,7 +57,7 @@ def is_subscribing_allowed():
Returns:
True if the limit is not reached, False otherwise
"""
return len(glob.glob("/etc/yunohost/dyndns/*.key"))<MAX_DYNDNS_DOMAINS
return len(glob.glob("/etc/yunohost/dyndns/*.key")) < MAX_DYNDNS_DOMAINS
def _dyndns_available(domain):
@ -101,9 +101,11 @@ def dyndns_subscribe(operation_logger, domain=None, key=None, password=None):
else:
from yunohost.utils.password import assert_password_is_strong_enough
# Ensure sufficiently complex password
if Moulinette.interface.type == "cli" and password==0:
if Moulinette.interface.type == "cli" and password == 0:
password = Moulinette.prompt(
m18n.n("ask_password"), is_password=True, confirm=True
m18n.n("ask_password"),
is_password=True,
confirm=True
)
operation_logger.data_to_redact.append(password)
assert_password_is_strong_enough("admin", password)
@ -162,8 +164,8 @@ def dyndns_subscribe(operation_logger, domain=None, key=None, password=None):
# Yeah the secret is already a base64-encoded but we double-bas64-encode it, whatever...
b64encoded_key = base64.b64encode(secret.encode()).decode()
data = {"subdomain": domain}
if password!=None:
data["recovery_password"]=hashlib.sha256((domain+":"+password.strip()).encode('utf-8')).hexdigest()
if password is not None:
data["recovery_password"] = hashlib.sha256((domain + ":" + password.strip()).encode('utf-8')).hexdigest()
r = requests.post(
f"https://{DYNDNS_PROVIDER}/key/{b64encoded_key}?key_algo=hmac-sha512",
data=data,
@ -182,7 +184,7 @@ def dyndns_subscribe(operation_logger, domain=None, key=None, password=None):
# Set the domain's config to autopush
from yunohost.domain import domain_config_set
domain_config_set(domain,key="dns.zone.autopush",value=1)
domain_config_set(domain, key="dns.zone.autopush", value=1)
# Yunohost regen conf will add the dyndns cron job if a key exists
# in /etc/yunohost/dyndns
@ -215,7 +217,8 @@ def dyndns_unsubscribe(operation_logger, domain, password=None):
# Ensure sufficiently complex password
if Moulinette.interface.type == "cli" and not password:
password = Moulinette.prompt(
m18n.n("ask_password"), is_password=True
m18n.n("ask_password"),
is_password=True
)
operation_logger.data_to_redact.append(password)
assert_password_is_strong_enough("admin", password)
@ -228,29 +231,30 @@ def dyndns_unsubscribe(operation_logger, domain, password=None):
import requests # lazy loading this module for performance reasons
# Send delete request
# Send delete request
try:
secret = str(domain) + ":" + str(password).strip()
r = requests.delete(
f"https://{DYNDNS_PROVIDER}/domains/{domain}",
data = {"recovery_password": hashlib.sha256(secret.encode('utf-8')).hexdigest()},
data={"recovery_password": hashlib.sha256(secret.encode('utf-8')).hexdigest()},
timeout=30,
)
except Exception as e:
raise YunohostError("dyndns_unregistration_failed", error=str(e))
if r.status_code == 200: # Deletion was successful
if r.status_code == 200: # Deletion was successful
rm(key_file, force=True)
# Yunohost regen conf will add the dyndns cron job if a key exists
# in /etc/yunohost/dyndns
regen_conf(["yunohost"])
logger.success(m18n.n("dyndns_unregistered"))
elif r.status_code == 403: # Wrong password
elif r.status_code == 403: # Wrong password
raise YunohostError("dyndns_unsubscribe_wrong_password")
elif r.status_code == 404: # Invalid domain
elif r.status_code == 404: # Invalid domain
raise YunohostError("dyndns_unsubscribe_wrong_domain")
def dyndns_list():
"""
Returns all currently subscribed DynDNS domains ( deduced from the key files )
@ -259,10 +263,11 @@ def dyndns_list():
files = glob.glob("/etc/yunohost/dyndns/K*key")
# Get the domain names
for i in range(len(files)):
files[i] = files[i].split(".+",1)[0]
files[i] = files[i].split(".+", 1)[0]
files[i] = files[i].split("/etc/yunohost/dyndns/K")[1]
return {"domains":files}
return {"domains": files}
@is_unit_operation()
def dyndns_update(
@ -289,17 +294,17 @@ def dyndns_update(
from yunohost.domain import domain_list
domains = domain_list(exclude_subdomains=True,auto_push=True)["domains"]
domains = domain_list(exclude_subdomains=True, auto_push=True)["domains"]
pushed = 0
for d in domains:
if is_yunohost_dyndns_domain(d):
dyndns_update(d, force=force, dry_run=dry_run)
pushed+=1
if pushed==0:
pushed += 1
if pushed == 0:
raise YunohostValidationError("dyndns_no_domain_registered")
return
elif type(domain).__name__ in ["list","tuple"]:
elif type(domain).__name__ in ["list", "tuple"]:
for d in domain:
dyndns_update(d, force=force, dry_run=dry_run)
return

View file

@ -18,7 +18,7 @@ from yunohost.domain import (
)
TEST_DOMAINS = ["example.tld", "sub.example.tld", "other-example.com"]
TEST_DYNDNS_DOMAIN = "".join(chr(random.randint(ord("a"),ord("z"))) for x in range(15))+random.choice([".noho.st",".ynh.fr",".nohost.me"])
TEST_DYNDNS_DOMAIN = "".join(chr(random.randint(ord("a"), ord("z"))) for x in range(15)) + random.choice([".noho.st", ".ynh.fr", ".nohost.me"])
TEST_DYNDNS_PASSWORD = "astrongandcomplicatedpassphrasethatisverysecure"
@ -41,7 +41,7 @@ def setup_function(function):
for domain in domains:
if (domain not in TEST_DOMAINS or domain == TEST_DOMAINS[2]) and domain != TEST_DYNDNS_DOMAIN:
# Clean domains not used for testing
domain_remove(domain,no_unsubscribe=is_yunohost_dyndns_domain(domain))
domain_remove(domain, no_unsubscribe=is_yunohost_dyndns_domain(domain))
elif domain in TEST_DOMAINS:
# Reset settings if any
os.system(f"rm -rf {DOMAIN_SETTINGS_DIR}/{domain}.yml")
@ -73,7 +73,7 @@ def test_domain_add():
def test_domain_add_subscribe():
assert TEST_DYNDNS_DOMAIN not in domain_list()["domains"]
domain_add(TEST_DYNDNS_DOMAIN,subscribe=TEST_DYNDNS_PASSWORD)
domain_add(TEST_DYNDNS_DOMAIN, subscribe=TEST_DYNDNS_PASSWORD)
assert TEST_DYNDNS_DOMAIN in domain_list()["domains"]
@ -91,7 +91,7 @@ def test_domain_remove():
def test_domain_remove_unsubscribe():
assert TEST_DYNDNS_DOMAIN in domain_list()["domains"]
domain_remove(TEST_DYNDNS_DOMAIN,unsubscribe=TEST_DYNDNS_PASSWORD)
domain_remove(TEST_DYNDNS_DOMAIN, unsubscribe=TEST_DYNDNS_PASSWORD)
assert TEST_DYNDNS_DOMAIN not in domain_list()["domains"]

View file

@ -233,7 +233,7 @@ def tools_postinstall(
# If this is a nohost.me/noho.st, actually check for availability
if is_yunohost_dyndns_domain(domain):
if ((subscribe==None) == (no_subscribe==False)):
if ((subscribe is None) == (no_subscribe is False)):
raise YunohostValidationError("domain_dyndns_instruction_unclear")
# Check if the domain is available...
@ -258,7 +258,7 @@ def tools_postinstall(
logger.info(m18n.n("yunohost_installing"))
# New domain config
domain_add(domain, subscribe=subscribe,no_subscribe=no_subscribe)
domain_add(domain, subscribe=subscribe, no_subscribe=no_subscribe)
domain_main_domain(domain)
# Update LDAP admin and create home dir

View file

@ -49,6 +49,7 @@ from yunohost.log import OperationLogger
logger = getActionLogger("yunohost.config")
CONFIG_PANEL_VERSION_SUPPORTED = 1.0
# Those js-like evaluate functions are used to eval safely visible attributes
# The goal is to evaluate in the same way than js simple-evaluate
# https://github.com/shepherdwind/simple-evaluate