auth/portal/acl : add 'user is allowed for domain X' mechanism, such that users can't log in or add mail aliases for a domain they aint allowed to access. The fact that they are able to access a domain is derived from the fact that they have access to at least one app on that domain (actually .. we may want to bypass this check for admins, otherwise this is gonna be hella confusing for fresh intalls).

This commit is contained in:
Alexandre Aubin 2023-12-21 18:36:15 +01:00
parent 20d914fd10
commit d0f1d9201c
4 changed files with 49 additions and 4 deletions

View file

@ -572,6 +572,7 @@
"log_user_permission_update": "Update accesses for permission '{}'",
"log_user_update": "Update info for user '{}'",
"mail_alias_remove_failed": "Could not remove e-mail alias '{mail}'",
"mail_alias_unauthorized": "You are not authorized to add aliases related to domain '{domain}'",
"mail_already_exists": "Mail adress '{mail}' already exists",
"mail_domain_unknown": "Invalid e-mail address for domain '{domain}'. Please, use a domain administrated by this server.",
"mail_forward_remove_failed": "Could not remove e-mail forwarding '{mail}'",

View file

@ -1723,7 +1723,7 @@ def app_ssowatconf():
for domain, apps in portal_domains_apps.items():
portal_settings = {}
portal_settings_path = Path(f"{PORTAL_SETTINGS_DIR}/{domain}.json")
portal_settings_path = Path(PORTAL_SETTINGS_DIR) / f"{domain}.json"
if portal_settings_path.exists():
portal_settings.update(read_json(str(portal_settings_path)))
@ -1735,6 +1735,13 @@ def app_ssowatconf():
str(portal_settings_path), portal_settings, sort_keys=True, indent=4
)
# Cleanup old files from possibly old domains
for setting_file in Path(PORTAL_SETTINGS_DIR).iterdir():
if setting_file.name.endswith(".json"):
domain = setting_file.name[:-len(".json")]
if domain not in portal_domains_apps:
setting_file.unlink()
logger.debug(m18n.n("ssowat_conf_generated"))

View file

@ -18,6 +18,7 @@ from cryptography.hazmat.backends import default_backend
from moulinette import m18n
from moulinette.authentication import BaseAuthenticator
from moulinette.utils.text import random_ascii
from moulinette.utils.filesystem import read_json
from yunohost.utils.error import YunohostError, YunohostAuthenticationError
logger = logging.getLogger("yunohostportal.authenticators.ldap_ynhuser")
@ -29,6 +30,34 @@ SESSION_VALIDITY = 3 * 24 * 3600 # 3 days
URI = "ldap://localhost:389"
USERDN = "uid={username},ou=users,dc=yunohost,dc=org"
DOMAIN_USER_ACL_DICT: dict[str, dict] = {}
PORTAL_SETTINGS_DIR = "/etc/yunohost/portal"
def user_is_allowed_on_domain(user: str, domain: str) -> bool:
assert "/" not in domain
portal_settings_path = Path(PORTAL_SETTINGS_DIR) / f"{domain}.json"
if not portal_settings_path.exists():
if "." not in domain:
return False
else:
parent_domain = domain.split(".", 1)[-1]
return user_is_allowed_on_domain(user, domain)
ctime = portal_settings_path.stat().st_ctime
if domain not in DOMAIN_USER_ACL_DICT or DOMAIN_USER_ACL_DICT[domain]["ctime"] < time.time():
users = set()
for infos in read_json(str(portal_settings_path))["apps"].values():
users = users.union(infos["users"])
DOMAIN_USER_ACL_DICT[domain] = {}
DOMAIN_USER_ACL_DICT[domain]["ctime"] = ctime
DOMAIN_USER_ACL_DICT[domain]["users"] = users
return user in DOMAIN_USER_ACL_DICT[domain]["users"]
# We want to save the password in the cookie, but we should do so in an encrypted fashion
# This is needed because the SSO later needs to possibly inject the Basic Auth header
@ -77,6 +106,8 @@ class Authenticator(BaseAuthenticator):
name = "ldap_ynhuser"
def _authenticate_credentials(self, credentials=None):
from bottle import request
try:
username, password = credentials.split(":", 1)
except ValueError:
@ -113,6 +144,9 @@ class Authenticator(BaseAuthenticator):
if con:
con.unbind_s()
if not user_is_allowed_on_domain(username, request.get_header("host")):
raise YunohostAuthenticationError("unable_authenticate")
return {"user": username, "pwd": encrypt(password)}
def set_session_cookie(self, infos):
@ -165,6 +199,9 @@ class Authenticator(BaseAuthenticator):
if infos["host"] != request.get_header("host"):
raise YunohostAuthenticationError("unable_authenticate")
if not user_is_allowed_on_domain(infos["user"], infos["host"]):
raise YunohostAuthenticationError("unable_authenticate")
self.purge_expired_session_files()
session_file = Path(SESSION_FOLDER) / infos["id"]
if not session_file.exists():

View file

@ -24,7 +24,7 @@ from typing import Any, Union
import ldap
from moulinette.utils.filesystem import read_json
from yunohost.authenticators.ldap_ynhuser import URI, USERDN, Authenticator as Auth
from yunohost.authenticators.ldap_ynhuser import URI, USERDN, Authenticator as Auth, user_is_allowed_on_domain
from yunohost.user import _hash_user_password
from yunohost.utils.error import YunohostError, YunohostValidationError
from yunohost.utils.ldap import LDAPInterface, _ldap_path_extract
@ -204,9 +204,9 @@ def portal_update(
"mail_already_exists", mail=mail, path=f"mailalias[{index}]"
)
if domain not in domains:
if domain not in domains or not user_is_allowed_on_domain(username, domain):
raise YunohostValidationError(
"mail_domain_unknown", domain=domain, path=f"mailalias[{index}]"
"mail_alias_unauthorized", domain=domain
)
mails.append(mail)