source: ogAgent-Git/src/opengnsys/modules/server/OpenGnSys/__init__.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#
# Copyright (c) 2014 Virtual Cable S.L.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without modification,
# are permitted provided that the following conditions are met:
#
#    * Redistributions of source code must retain the above copyright notice,
#      this list of conditions and the following disclaimer.
#    * Redistributions in binary form must reproduce the above copyright notice,
#      this list of conditions and the following disclaimer in the documentation
#      and/or other materials provided with the distribution.
#    * Neither the name of Virtual Cable S.L. nor the names of its contributors
#      may be used to endorse or promote products derived from this software
#      without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
# DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
# FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
# DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
# SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
# CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
# OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
"""
@author: Ramón M. Gómez, ramongomez at us dot es
"""

import base64
import os
import random
import shutil
import string
import threading
import time
import urllib.error
import urllib.parse
import urllib.request

from configparser import NoOptionError
from opengnsys import REST, operations, VERSION
from opengnsys.log import logger
from opengnsys.jobmgr import JobMgr
from opengnsys.workers import ServerWorker


# Check authorization header decorator
def check_secret(fnc):
    """
    Decorator to check for received secret key and raise exception if it isn't valid.
    """
    def wrapper(*args, **kwargs):
        try:
            this, path, get_params, post_params, server = args
            # Accept "status" operation with no arguments or any function with Authorization header
            if fnc.__name__ == 'process_status' and not get_params:
                return fnc(*args, **kwargs)
            elif this.random == server.headers['Authorization']:
                return fnc(*args, **kwargs)
            else:
                raise Exception('Unauthorized operation')
        except Exception as e:
            logger.debug (str(e))
            raise Exception(e)

    return wrapper


# Check if operation is permitted
def execution_level(level):
    def check_permitted(fnc):
        def wrapper(*args, **kwargs):
            levels = ['status', 'halt', 'full']
            this = args[0]
            try:
                if levels.index(level) <= levels.index(this.exec_level):
                    return fnc(*args, **kwargs)
                else:
                    raise Exception('Unauthorized operation')
            except Exception as e:
                logger.debug (str(e))
                raise Exception(e)

        return wrapper

    return check_permitted


class OpenGnSysWorker(ServerWorker):
    name = 'opengnsys'  # Module name
    interface = None  # Bound interface for OpenGnsys
    REST = None  # REST object
    user = []  # User sessions
    session_type = ''  # User session type
    random = None  # Random string for secure connections
    length = 32  # Random string length
    exec_level = None  # Execution level (permitted operations)
    jobmgr = JobMgr()

    ## pings ogcore
    def mon (self):
        n = 0
        while True:
            time.sleep (1)
            n += 1
            if not n % 10:
                body = {
                    "iph": self.interface.ip,
                    "timestamp": int (time.time()),
                }
                #logger.debug (f'about to send ping ({body})')
                self.REST.sendMessage ('clients/status/webhook', body)

    def onActivation(self):
        """
        Sends OGAgent activation notification to OpenGnsys server
        """
        if os.path.exists ('/scripts/oginit'):
            ## estamos en oglive, este modulo no debe cargarse
            ## esta lógica la saco de src/opengnsys/linux/operations.py, donde hay un if similar
            raise Exception ('Refusing to load within an ogLive image')

        e = None  # Error info
        t = 0  # Count of time
        # Generate random secret to send on activation
        self.random = ''.join(random.choice(string.ascii_lowercase + string.digits) for _ in range(self.length))
        # Ensure cfg has required configuration variables or an exception will be thrown
        try:
            url      = self.service.config.get(self.name, 'remote')
            ca_file  = self.service.config.get(self.name, 'ca')
            crt_file = self.service.config.get(self.name, 'crt')
            key_file = self.service.config.get(self.name, 'key')
        except NoOptionError as e:
            logger.error("Configuration error: {}".format(e))
            raise e
        self.REST = REST (url, ca_file=ca_file, crt_file=crt_file, key_file=key_file)
        # Execution level ('full' by default)
        try:
            self.exec_level = self.service.config.get(self.name, 'level')
        except NoOptionError:
            self.exec_level = 'full'
        # Get network interfaces until they are active or timeout (5 minutes)
        for t in range(0, 300):
            try:
                # Get the first network interface
                nets = list (operations.getNetworkInfo())
                if 0 == len (nets):
                    logger.error ('No network interfaces found')
                    raise Exception ('No network interfaces found')
                self.interface = nets[0]
            except Exception as e:
                # Wait 1 sec. and retry
                logger.warn (e)
                time.sleep(1)
            finally:
                # Exit loop if interface is active
                if self.interface:
                    if t > 0:
                        logger.debug("Fetch connection data after {} tries".format(t))
                    break
        # Raise error after timeout
        if not self.interface:
            ## UnboundLocalError: cannot access local variable 'e' where it is not associated with a value
            raise e

        # Loop to send initialization message
        init_retries = 100
        for t in range(0, init_retries):
            try:
                try:
                    self.REST.sendMessage('ogagent/started', {'mac': self.interface.mac, 'ip': self.interface.ip,
                                                              'secret': self.random, 'ostype': operations.os_type,
                                                              'osversion': operations.os_version,
                                                              'agent_version': VERSION})
                    break
                except Exception as e:
                    logger.warn (str (e))
                    # Trying to initialize on alternative server, if defined
                    # (used in "exam mode" from the University of Seville)
                    self.REST = REST(self.service.config.get(self.name, 'altremote'), ca_file=ca_file, crt_file=crt_file, key_file=key_file)
                    self.REST.sendMessage('ogagent/started', {'mac': self.interface.mac, 'ip': self.interface.ip,
                                                              'secret': self.random, 'ostype': operations.os_type,
                                                              'osversion': operations.os_version, 'alt_url': True,
                                                              'agent_version': VERSION})
                    break
            except Exception as e:
                logger.warn (str (e))
                time.sleep(3)
        # Raise error after timeout
        if t < init_retries-1:
            logger.debug('Successful connection after {} tries'.format(t))
        elif t == init_retries-1:
            raise Exception('Initialization error: Cannot connect to remote server')

        # Delete marking files
        for f in ['ogboot.me', 'ogboot.firstboot', 'ogboot.secondboot']:
            try:
                os.remove(os.sep + f)
            except OSError:
                pass
        # Copy file "HostsFile.FirstOctetOfIPAddress" to "HostsFile", if it exists
        # (used in "exam mode" from the University of Seville)
        hosts_file = os.path.join(operations.get_etc_path(), 'hosts')
        new_hosts_file = hosts_file + '.' + self.interface.ip.split('.')[0]
        if os.path.isfile(new_hosts_file):
            shutil.copyfile(new_hosts_file, hosts_file)

        threading.Thread (name='monitoring_thread', target=self.mon, daemon=True).start()

        logger.debug ('onActivation ok')

    def onDeactivation(self):
        """
        Sends OGAgent stopping notification to OpenGnsys server
        """
        now = time.time()
        for elem in self.user:
            sess_len = now - elem['login_ts']
            logger.debug ('Session of logged in user {} took {} seconds'.format (elem['username'], int (sess_len)))
        logger.debug('onDeactivation')
        self.REST.sendMessage('ogagent/stopped', {'mac': self.interface.mac, 'ip': self.interface.ip,
                                                  'ostype': operations.os_type, 'osversion': operations.os_version})

    def onLogin(self, data):
        """
        Sends session login notification to OpenGnsys server
        """
        user, language, self.session_type = tuple(data.split(','))
        logger.debug('Received login for {0} using {2} with language {1}'.format(user, language, self.session_type))
        self.user.append ({'username': user, 'login_ts': time.time() })
        self.REST.sendMessage('ogagent/loggedin', {'ip': self.interface.ip, 'user': user, 'language': language,
                                                   'session': self.session_type,
                                                   'ostype': operations.os_type, 'osversion': operations.os_version})

    def onLogout(self, user):
        """
        Sends session logout notification to OpenGnsys server
        """
        sess_len = 0
        for elem in self.user:
            if user != elem['username']: continue
            sess_len = time.time() - elem['login_ts']
        logger.debug ('Received logout for {}, session length {} seconds'.format (user, int (sess_len)))
        try:
            self.user.pop()
        except IndexError:
            pass
        self.REST.sendMessage('ogagent/loggedout', {'ip': self.interface.ip, 'user': user})

    def process_ogclient(self, path, get_params, post_params, server):
        """
        This method can be overridden to provide your own message processor, or better you can
        implement a method that is called exactly as "process_" + path[0] (module name has been removed from path
        array) and this default processMessage will invoke it
        * Example:
            Imagine this invocation url (no matter if GET or POST): http://example.com:9999/Sample/mazinger/Z
            The HTTP Server will remove "Sample" from path, parse arguments and invoke this method as this:
            module.processMessage(["mazinger","Z"], get_params, post_params)

            This method will process "mazinger", and look for a "self" method that is called "process_mazinger",
            and invoke it this way:
               return self.process_mazinger(["Z"], get_params, post_params)

            In the case path is empty (that is, the path is composed only by the module name, like in
            "http://example.com/Sample", the "process" method will be invoked directly

            The methods must return data that can be serialized to json (i.e. Objects are not serializable to json,
            basic type are)
        """
        if not path:
            return "ok"
        try:
            operation = getattr(self, 'ogclient_' + path[0])
        except Exception:
            raise Exception('Message processor for "{}" not found'.format(path[0]))
        return operation(path[1:], get_params, post_params)

    # Warning: the order of the decorators matters
    @execution_level('status')
    @check_secret
    def process_status(self, path, get_params, post_params, server):
        """
        Returns client status (OS type or execution status) and login status
        :param path:
        :param get_params: optional parameter "detail" to show extended status
        :param post_params:
        :param server:
        :return: JSON object {"status": "status_code", "loggedin": boolean, ...}
        """
        st = {'linux': 'LNX', 'macos': 'OSX', 'windows': 'WIN'}
        try:
            # Standard status
            res = {'status': st[operations.os_type.lower()], 'loggedin': len(self.user) > 0,
                   'session': self.session_type}
            # Detailed status
            if get_params.get('detail', 'false') == 'true':
                res.update({'agent_version': VERSION, 'os_version': operations.os_version, 'sys_load': os.getloadavg()})
                if res['loggedin']:
                    res.update({'sessions': len(self.user), 'current_user': self.user[-1]['username']})
        except KeyError:
            # Unknown operating system
            res = {'status': 'UNK'}
        return res

    @execution_level('halt')
    @check_secret
    def process_Reiniciar(self, path, get_params, post_params, server):
        """
        Launches a system reboot operation
        :param path:
        :param get_params:
        :param post_params:
        :param server: authorization header
        :return: JSON object {"op": "launched"}
        """
        logger.debug('Received reboot operation')

        # Rebooting thread
        def rebt():
            operations.reboot()
        threading.Thread(target=rebt).start()
        return {'op': 'launched'}

    @execution_level('halt')
    @check_secret
    def process_Apagar(self, path, get_params, post_params, server):
        """
        Launches a system power off operation
        :param path:
        :param get_params:
        :param post_params:
        :param server: authorization header
        :return: JSON object {"op": "launched"}
        """
        logger.debug('Received poweroff operation')

        # Powering off thread
        def pwoff():
            time.sleep(2)
            operations.poweroff()
        threading.Thread(target=pwoff).start()
        return {'op': 'launched'}

    @execution_level('full')
    @check_secret
    def process_EjecutarScript(self, path, get_params, post_params, server):
        """
        Processes an script execution (script should be encoded in base64)
        :param path:
        :param get_params:
        :param post_params: JSON object {"script": "commands"}
        :param server: authorization header
        :return: JSON object {"op": "launched"}
        """
        logger.debug('Processing script request')
        # Decoding script
        param_script = post_params.get('script')
        if not param_script:
            return {'op': 'error', 'err': 'Required parameter "script" is missing or empty'}
        try:
            b64decoded = base64.b64decode (param_script)
        except Exception as e:
            return {'op': 'error', 'err': f'Failed to decode base64: {e}'}
        script = urllib.parse.unquote (b64decoded.decode ('utf-8'))
        logger.debug('received script "{}"'.format(script))

        if post_params.get('client', 'false') == 'false':
            jobid = self.jobmgr.launch_job (script, False)
            return {'op': 'launched', 'jobid': jobid}

        else:   ## post_params.get('client') is not 'false'
            ## send script as-is
            self.sendClientMessage('script', {'code': script})
            #return {'op': 'launched', 'jobid': jobid}       ## TODO obtain jobid generated at the client (can it be done?)
            return {'op': 'launched'}

    @execution_level('full')
    @check_secret
    def process_terminatescript(self, path, get_params, post_params, server):
        jobid = post_params.get('jobid', None)
        logger.debug('Processing terminate_script request, jobid "{}"'.format (jobid))
        if jobid is None:
            return {}
        self.sendClientMessage('terminatescript', {'jobid': jobid})
        self.jobmgr.terminate_job (jobid)
        return {}

    @execution_level('full')
    @check_secret
    def process_preparescripts(self, path, get_params, post_params, server):
        logger.debug('Processing preparescripts request')
        self.st = self.jobmgr.prepare_jobs()
        logger.debug('Sending preparescripts to client')
        self.sendClientMessage('preparescripts', None)
        return {}

    def process_client_preparescripts(self, params):
        logger.debug('Processing preparescripts message from client')
        for p in params:
            #logger.debug ('p "{}"'.format(p))
            self.st.append (p)

    @execution_level('full')
    @check_secret
    def process_getscripts(self, path, get_params, post_params, server):
        logger.debug('Processing getscripts request')
        return self.st

    @execution_level('full')
    @check_secret
    def process_logoff(self, path, get_params, post_params, server):
        """
        Closes user session
        """
        logger.debug('Received logoff operation')
        # Sending log off message to OGAgent client
        self.sendClientMessage('logoff', {})
        return {'op': 'sent to client'}

    @execution_level('full')
    @check_secret
    def process_popup(self, path, get_params, post_params, server):
        """
        Shows a message popup on the user's session
        """
        logger.debug('Received message operation')
        # Sending popup message to OGAgent client
        self.sendClientMessage('popup', post_params)
        return {'op': 'launched'}

    def process_client_popup(self, params):
        self.REST.sendMessage('popup_done', params)