pyceo/ceod/model/CloudService.py

from base64 import b64encode
import datetime
import hashlib
import hmac
import ipaddress
import json
import os
import re
from typing import Dict, List
from urllib.parse import quote

import requests
from zope import component
from zope.interface import implementer

from .VHostManager import VHostManager
from ceo_common.errors import CloudStackAPIError, InvalidDomainError, InvalidIPError
from ceo_common.logger_factory import logger_factory
from ceo_common.interfaces import ICloudService, IConfig, IUser, ILDAPService, \
    IMailService
from ceo_common.model import Term
import ceo_common.utils as utils

logger = logger_factory(__name__)


@implementer(ICloudService)
class CloudService:
    VALID_DOMAIN_RE = re.compile(r'^(?:[0-9a-z-]+\.)+[a-z]+$')

    def __init__(self):
        cfg = component.getUtility(IConfig)
        self.api_key = cfg.get('cloudstack_api_key')
        self.secret_key = cfg.get('cloudstack_secret_key')
        self.base_url = cfg.get('cloudstack_base_url')
        self.members_domain = 'Members'
        self.vhost_mgr = VHostManager(
            vhost_dir=cfg.get('cloud vhosts_config_dir'),
            ssl_cert_path=cfg.get('cloud vhosts_ssl_cert_path'),
            ssl_key_path=cfg.get('cloud vhosts_ssl_key_path'),
        )
        self.max_vhosts_per_account = cfg.get('cloud vhosts_max_vhosts_per_account')
        self.vhost_domain = cfg.get('cloud vhosts_members_domain')
        self.vhost_ip_min = ipaddress.ip_address(cfg.get('cloud vhosts_ip_range_min'))
        self.vhost_ip_max = ipaddress.ip_address(cfg.get('cloud vhosts_ip_range_max'))

        state_dir = '/run/ceod'
        if not os.path.isdir(state_dir):
            os.mkdir(state_dir)
        self.pending_deletions_file = os.path.join(state_dir, 'cloudstack_pending_account_deletions.json')

    def _create_url(self, params: Dict[str, str]) -> str:
        # See https://docs.cloudstack.apache.org/en/latest/developersguide/dev.html#the-cloudstack-api
        if 'apiKey' not in params and 'apikey' not in params:
            params['apiKey'] = self.api_key
        params['response'] = 'json'
        request_str = '&'.join(
            key + '=' + quote(val)
            for key, val in params.items()
        )
        sig_str = '&'.join(
            key.lower() + '=' + quote(val).lower()
            for key, val in sorted(params.items())
        )
        sig = hmac.new(self.secret_key.encode(), sig_str.encode(), hashlib.sha1).digest()
        encoded_sig = b64encode(sig).decode()
        url = self.base_url + '?' + request_str + '&signature=' + quote(encoded_sig)
        return url

    def _get_domain_id(self, domain_name: str) -> str:
        url = self._create_url({
            'command': 'listDomains',
            'details': 'min',
            'name': domain_name,
        })
        resp = requests.get(url)
        resp.raise_for_status()
        d = resp.json()['listdomainsresponse']
        assert d['count'] == 1, 'there should be one domain found'
        return d['domain'][0]['id']

    def _get_all_accounts(self, domain_id: str) -> List[Dict]:
        url = self._create_url({
            'command': 'listAccounts',
            'domainid': domain_id,
            'details': 'min',
        })
        resp = requests.get(url)
        resp.raise_for_status()
        d = resp.json()['listaccountsresponse']
        if 'account' not in d:
            # The API returns an empty dict if there are no accounts
            return []
        return d['account']

    def _delete_account(self, account_id: str):
        url = self._create_url({
            'command': 'deleteAccount',
            'id': account_id,
        })
        resp = requests.post(url)
        resp.raise_for_status()

    def create_account(self, user: IUser):
        domain_id = self._get_domain_id(self.members_domain)

        url = self._create_url({
            'command': 'ldapCreateAccount',
            'accounttype': '0',
            'domainid': domain_id,
            'username': user.uid,
        })
        resp = requests.post(url)
        d = resp.json()['createaccountresponse']
        if not resp.ok:
            raise CloudStackAPIError(d['errortext'])

    def purge_accounts(self) -> Dict:
        accounts_deleted = []
        accounts_to_be_deleted = []
        result = {
            'accounts_deleted': accounts_deleted,
            'accounts_to_be_deleted': accounts_to_be_deleted,
        }

        current_term = Term.current()
        beginning_of_term = current_term.to_datetime()
        now = utils.get_current_datetime()
        delta = now - beginning_of_term
        if delta.days < 30:
            # one-month grace period
            return result

        ldap_srv = component.getUtility(ILDAPService)
        mail_srv = component.getUtility(IMailService)
        domain_id = self._get_domain_id(self.members_domain)
        accounts = self._get_all_accounts(domain_id)

        if os.path.isfile(self.pending_deletions_file):
            state = json.load(open(self.pending_deletions_file))
            last_check = datetime.datetime.fromtimestamp(state['timestamp'])
            delta = now - last_check
            if delta.days < 7:
                logger.debug(
                    'Skipping account purge because less than one week has '
                    'passed since the warning emails were sent out'
                )
                accounts_to_be_deleted.extend(state['accounts_to_be_deleted'])
                return result
            username_to_account_id = {
                account['name']: account['id']
                for account in accounts
            }
            for username in state['accounts_to_be_deleted']:
                if username not in username_to_account_id:
                    continue
                user = ldap_srv.get_user(username)
                if user.membership_is_valid():
                    continue
                account_id = username_to_account_id[username]

                self._delete_account(account_id)
                self.vhost_mgr.delete_all_vhosts_for_user(username)

                accounts_deleted.append(username)
                mail_srv.send_cloud_account_has_been_deleted_message(user)
                logger.info(f'Deleted cloud account for {username}')
            os.unlink(self.pending_deletions_file)
            return result

        state = {
            'timestamp': int(now.timestamp()),
            'accounts_to_be_deleted': accounts_to_be_deleted,
        }
        for account in accounts:
            username = account['name']
            account_id = account['id']
            user = ldap_srv.get_user(username)
            if user.membership_is_valid():
                continue
            accounts_to_be_deleted.append(username)
            mail_srv.send_cloud_account_will_be_deleted_message(user)
            logger.info(
                f'A warning email was sent to {username} because their '
                'cloud account will be deleted'
            )
        if accounts_to_be_deleted:
            json.dump(state, open(self.pending_deletions_file, 'w'))
        return result

    def _is_valid_domain(self, username: str, domain: str) -> bool:
        subdomain = username + '.' + self.vhost_domain
        if not (domain == subdomain or domain.endswith('.' + subdomain)):
            return False
        if self.VALID_DOMAIN_RE.match(domain) is None:
            return False
        if len(domain) > 80:
            return False
        return True

    def _is_valid_ip_address(self, ip_address: str) -> bool:
        try:
            addr = ipaddress.ip_address(ip_address)
        except ValueError:
            return False
        return self.vhost_ip_min <= addr <= self.vhost_ip_max

    def create_vhost(self, username: str, domain: str, ip_address: str):
        if self.vhost_mgr.get_num_vhosts(username) >= self.max_vhosts_per_account:
            raise Exception(f'Only {self.max_vhosts_per_account} vhosts '
                            'allowed per account')
        if not self._is_valid_domain(username, domain):
            raise InvalidDomainError()
        if not self._is_valid_ip_address(ip_address):
            raise InvalidIPError()
        self.vhost_mgr.create_vhost(username, domain, ip_address)

    def delete_vhost(self, username: str, domain: str):
        if not self._is_valid_domain(username, domain):
            raise InvalidDomainError()
        self.vhost_mgr.delete_vhost(username, domain)

    def get_vhosts(self, username: str) -> List[Dict]:
        return self.vhost_mgr.get_vhosts(username)