src/portal.py

# -*- coding: utf-8 -*-

""" License

    Copyright (C) 2021 YUNOHOST.ORG

    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU Affero General Public License as published
    by the Free Software Foundation, either version 3 of the License, or
    (at your option) any later version.

    This program is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
    GNU Affero General Public License for more details.

    You should have received a copy of the GNU Affero General Public License
    along with this program; if not, see http://www.gnu.org/licenses

"""
from pathlib import Path
from typing import Any, Union

import ldap
from moulinette.utils.filesystem import read_json, read_yaml
from moulinette.utils.log import getActionLogger
from yunohost.authenticators.ldap_ynhuser import URI, USERDN, Authenticator as Auth
from yunohost.user import _hash_user_password
from yunohost.utils.error import YunohostError, YunohostValidationError
from yunohost.utils.ldap import LDAPInterface
from yunohost.utils.password import (
    assert_password_is_compatible,
    assert_password_is_strong_enough,
)

logger = getActionLogger("portal")

ADMIN_ALIASES = ["root", "admin", "admins", "webmaster", "postmaster", "abuse"]


def _get_user_infos(
    user_attrs: list[str],
) -> tuple[str, str, dict[str, Any], LDAPInterface]:
    auth = Auth().get_session_cookie(decrypt_pwd=True)
    username = auth["user"]
    ldap_interface = LDAPInterface(username, auth["pwd"])
    result = ldap_interface.search("ou=users", f"uid={username}", user_attrs)
    if not result:
        raise YunohostValidationError("user_unknown", user=username)

    return username, auth["host"], result[0], ldap_interface


def _get_portal_settings(domain: Union[str, None] = None):
    from yunohost.domain import DOMAIN_SETTINGS_DIR

    if not domain:
        from bottle import request

        domain = request.get_header("host")

    assert domain and "/" not in domain

    settings = {
        "public": False,
        "portal_logo": "",
        "portal_theme": "system",
        "portal_title": "YunoHost",
        "show_other_domains_apps": False,
        "domain": domain,
    }

    if Path(f"{DOMAIN_SETTINGS_DIR}/{domain}.portal.yml").exists():
        settings.update(read_yaml(f"{DOMAIN_SETTINGS_DIR}/{domain}.portal.yml"))

    return settings


def portal_public():
    portal_settings = _get_portal_settings()
    portal_settings["apps"] = {}
    portal_settings["public"] = portal_settings.pop("default_app") == "_yunohost_portal_with_public_apps"

    if portal_settings["public"]:
        ssowat_conf = read_json("/etc/ssowat/conf.json")
        portal_settings["apps"] = {
            perm.replace(".main", ""): {
                "label": infos["label"],
                "url": infos["uris"][0],
            }
            for perm, infos in ssowat_conf["permissions"].items()
            if infos["show_tile"] and infos["public"]
        }

        if not portal_settings["show_other_domains_apps"]:
            portal_settings["apps"] = {
                name: data
                for name, data in portal_settings["apps"].items()
                if portal_settings["domain"] in data["url"]
            }

    return portal_settings


def portal_me():
    """
    Get user informations
    """
    username, domain, user, _ = _get_user_infos(
        ["cn", "mail", "maildrop", "mailuserquota", "memberOf", "permission"]
    )

    groups = [
        g.replace("cn=", "").replace(",ou=groups,dc=yunohost,dc=org", "")
        for g in user["memberOf"]
    ]
    groups = [g for g in groups if g not in [username, "all_users"]]

    permissions = [
        p.replace("cn=", "").replace(",ou=permission,dc=yunohost,dc=org", "")
        for p in user["permission"]
    ]

    ssowat_conf = read_json("/etc/ssowat/conf.json")
    apps = {
        perm.replace(".main", ""): {"label": infos["label"], "url": infos["uris"][0]}
        for perm, infos in ssowat_conf["permissions"].items()
        if perm in permissions and infos["show_tile"] and username in infos["users"]
    }

    settings = _get_portal_settings(domain=domain)
    if not settings["show_other_domains_apps"]:
        apps = {name: data for name, data in apps.items() if domain in data["url"]}

    result_dict = {
        "username": username,
        "fullname": user["cn"][0],
        "mail": user["mail"][0],
        "mailalias": user["mail"][1:],
        "mailforward": user["maildrop"][1:],
        "groups": groups,
        "apps": apps,
    }

    # FIXME / TODO : add mail quota status ?
    #  result_dict["mailbox-quota"] = {
    #      "limit": userquota if is_limited else m18n.n("unlimit"),
    #      "use": storage_use,
    #  }
    # Could use : doveadm -c /dev/null -f flow quota recalc -u johndoe
    # But this requires to be in the mail group ...

    return result_dict


def portal_update(
    fullname: Union[str, None] = None,
    mailforward: Union[list[str], None] = None,
    mailalias: Union[list[str], None] = None,
    currentpassword: Union[str, None] = None,
    newpassword: Union[str, None] = None,
):
    from yunohost.domain import domain_list

    domains = domain_list()["domains"]
    username, domain, current_user, ldap_interface = _get_user_infos(
        ["givenName", "sn", "cn", "mail", "maildrop", "memberOf"]
    )
    new_attr_dict = {}

    if fullname is not None and fullname != current_user["cn"]:
        fullname = fullname.strip()
        firstname = fullname.split()[0]
        lastname = (
            " ".join(fullname.split()[1:]) or " "
        )  # Stupid hack because LDAP requires the sn/lastname attr, but it accepts a single whitespace...
        new_attr_dict["givenName"] = [firstname]  # TODO: Validate
        new_attr_dict["sn"] = [lastname]  # TODO: Validate
        new_attr_dict["cn"] = new_attr_dict["displayName"] = [
            (firstname + " " + lastname).strip()
        ]

    if mailalias is not None:
        mailalias = [mail.strip() for mail in mailalias if mail and mail.strip()]
        # keep first current mail unaltered
        mails = [current_user["mail"][0]]

        for index, mail in enumerate(mailalias):
            if mail in current_user["mail"]:
                if mail != current_user["mail"][0] and mail not in mails:
                    mails.append(mail)
                continue  # already in mails, skip validation

            local_part, domain = mail.split("@")
            if local_part in ADMIN_ALIASES:
                raise YunohostValidationError(
                    "mail_unavailable", path=f"mailalias[{index}]"
                )

            try:
                ldap_interface.validate_uniqueness({"mail": mail})
            except Exception as e:
                raise YunohostError("user_update_failed", user=username, error=e)

            if domain not in domains:
                raise YunohostValidationError(
                    "mail_domain_unknown", domain=domain, path=f"mailalias[{index}]"
                )

            mails.append(mail)

        new_attr_dict["mail"] = mails

    if mailforward is not None:
        new_attr_dict["maildrop"] = [current_user["maildrop"][0]] + [
            mail.strip()
            for mail in mailforward
            if mail and mail.strip() and mail != current_user["maildrop"][0]
        ]

    if newpassword:
        # Check that current password is valid
        try:
            con = ldap.ldapobject.ReconnectLDAPObject(URI, retry_max=0)
            con.simple_bind_s(USERDN.format(username=username), currentpassword)
        except ldap.INVALID_CREDENTIALS:
            raise YunohostValidationError("invalid_password", path="currentpassword")
        finally:
            # Free the connection, we don't really need it to keep it open as the point is only to check authentication...
            if con:
                con.unbind_s()

        # Ensure compatibility and sufficiently complex password
        try:
            assert_password_is_compatible(newpassword)
            is_admin = (
                "cn=admins,ou=groups,dc=yunohost,dc=org" in current_user["memberOf"]
            )
            assert_password_is_strong_enough(
                "admin" if is_admin else "user", newpassword
            )
        except YunohostValidationError as e:
            raise YunohostValidationError(e.key, path="newpassword")

        Auth().delete_session_cookie()
        new_attr_dict["userPassword"] = [_hash_user_password(newpassword)]

    try:
        ldap_interface.update(f"uid={username},ou=users", new_attr_dict)
    except Exception as e:
        raise YunohostError("user_update_failed", user=username, error=e)

    # FIXME: Here we could want to trigger "post_user_update" hook but hooks has to
    # be run as root

    return {
        "fullname": new_attr_dict["cn"][0],
        "mailalias": new_attr_dict["mail"][1:],
        "mailforward": new_attr_dict["maildrop"][1:],
    }