pyceo/ceod/model/CloudService.py

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

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

from ceo_common.errors import InvalidMembershipError, CloudStackAPIError
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:
    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'

        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):
        if not user.membership_is_valid():
            raise InvalidMembershipError()
        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)
                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