Advancing on Linux Actor with very similar structure to windows actor

actors/src/test.py

@@ -54,6 +54,13 @@ def ipcTest():
     client2 = ipc.ClientIPC(39188)
     client2.start()
 
+    print("Requesting information")
+    client.requestInformation()
+    print("Sending login info")
+    client.sendLogin('user1')
+    print("Sending logout info")
+    client.sendLogout('mariete' * 1000)
+
     print('Sending message')
     s.sendMessage(ipc.MSG_LOGOFF, None)
     s.sendMessage(ipc.MSG_MESSAGE, 'Cierra la sesión')
@@ -209,8 +216,8 @@ def testRemote():
 
 if __name__ == '__main__':
     # ipcServer()
-    # ipcTest()
+    ipcTest()
-    testRest()
+    # testRest()
     # testIdle()
     # testServer()
     # testRemote()

actors/src/udsactor/ipc.py

@@ -25,7 +25,6 @@
 # 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: Adolfo Gómez, dkmaster at dkmon dot com
 '''
@@ -33,6 +32,7 @@ from __future__ import unicode_literals
 
 import socket
 import threading
+import sys
 import six
 import traceback
 import pickle
@@ -42,12 +42,18 @@ from udsactor.log import logger
 
 # The IPC Server will wait for connections from clients
 # Clients will open socket, and wait for data from server
-# The messages sent will be the following (subject to future changes):
+# The messages sent (from server) will be the following (subject to future changes):
 #     Message_id     Data               Action
 #    ------------  --------         --------------------------
 #    MSG_LOGOFF     None            Logout user from session
 #    MSG_MESSAGE    message,level   Display a message with level (INFO, WARN, ERROR, FATAL)     # TODO: Include level, right now only has message
 #    MSG_SCRIPT     python script   Execute an specific python script INSIDE CLIENT environment (this messages is not sent right now)
+# The messages received (sent from client) will be the following:
+#     Message_id       Data               Action
+#    ------------    --------         --------------------------
+#    REQ_LOGOUT                   Logout user from session
+#    REQ_INFORMATION  None            Request information from ipc server (maybe configuration parameters in a near future)
+#    REQ_LOGIN        python script   Execute an specific python script INSIDE CLIENT environment (this messages is not sent right now)
 #
 # All messages are in the form:
 # BYTE
@@ -58,7 +64,12 @@ from udsactor.log import logger
 MSG_LOGOFF = 0xA1
 MSG_MESSAGE = 0xB2
 MSG_SCRIPT = 0xC3
-MSG_INFORMATION = 0x90
+MSG_INFORMATION = 0xD4
+
+# Request messages
+REQ_INFORMATION = MSG_INFORMATION
+REQ_LOGIN = 0xE5
+REQ_LOGOUT = MSG_LOGOFF
 
 VALID_MESSAGES = (MSG_LOGOFF, MSG_MESSAGE, MSG_SCRIPT, MSG_INFORMATION)
 
@@ -67,6 +78,16 @@ REQ_INFORMATION = 0xAA
 MAGIC = b'\x55\x44\x53\x00'  # UDS in hexa with a padded 0 to the right
 
 
+# Allows notifying login/logout from client for linux platform
+ALLOW_LOG_METHODS = sys.platform != 'win32'
+
+
+# States for client processor
+ST_SECOND_BYTE = 0x01
+ST_RECEIVING = 0x02
+ST_PROCESS_MESSAGE = 0x02
+
+
 class ClientProcessor(threading.Thread):
     def __init__(self, parent, clientSocket):
         super(self.__class__, self).__init__()
@@ -79,30 +100,64 @@ class ClientProcessor(threading.Thread):
         logger.debug('Stoping client processor')
         self.running = False
 
+    def processRequest(self, msg, data):
+        print('Got message {}, with data {}'.format(msg, data))
+
     def run(self):
         self.running = True
         self.clientSocket.setblocking(0)
 
+        state = None
+        recv_msg = None
+        recv_data = None
         while self.running:
             try:
-                while True:
+                counter = 1024
-                    buf = bytearray(self.clientSocket.recv(512))  # Empty buffer, this is set as non-blocking
+                while counter > 0:  # So we process at least the incoming queue every XX bytes readed
-                    if len(buf) == 0:  # No data
+                    counter -= 1
+                    b = self.clientSocket.recv(1)
+                    if b == b'':
                         break
-                    for b in buf:
+                    buf = six.byte2int(b)  # Empty buffer, this is set as non-blocking
-                        if b == REQ_INFORMATION:
+                    if state is None:
-                            infoParams = self.parent.infoParams if self.parent.infoParams is not None else {}
+                        if buf in (REQ_INFORMATION, REQ_LOGIN, REQ_LOGOUT):
-                            self.messages.put((MSG_INFORMATION, pickle.dumps(infoParams)))
+                            print('State set to {}'.format(buf))
-                            logger.debug('Received a request for information')
+                            state = buf
+                            recv_msg = buf
+                            continue  # Get next byte
                         else:
-                            logger.debug('Got unexpected data {}'.format(ord(b)))
+                            logger.debug('Got unexpected data {}'.format(buf))
-                # In fact, we do not process anything right now, simply empty recv buffer if something is found
+                    elif state in (REQ_INFORMATION, REQ_LOGIN, REQ_LOGOUT):
+                        print('First length byte is {}'.format(buf))
+                        msg_len = buf
+                        state = ST_SECOND_BYTE
+                        continue
+                    elif state == ST_SECOND_BYTE:
+                        msg_len += buf << 8
+                        print('Second length byte is {}, len is {}'.format(buf, msg_len))
+                        if msg_len == 0:
+                            self.processRequest(recv_msg, None)
+                            state = None
+                            break
+                        state = ST_RECEIVING
+                        recv_data = b''
+                        continue
+                    elif state == ST_RECEIVING:
+                        recv_data += six.int2byte(buf)
+                        msg_len -= 1
+                        if msg_len == 0:
+                            self.processRequest(recv_msg, recv_data)
+                            recv_data = None
+                            state = None
+                            break
+                    else:
+                        logger.debug('Got invalid message from request: {}, state: {}'.format(buf, state))
             except socket.error as e:
-                # If no data is present, no problem at all
+                # If no data is present, no problem at all, pass to check messages
                 pass
 
             try:
-                msg = self.messages.get(block=True, timeout=1)
+                msg = self.messages.get(block=False)
             except six.moves.queue.Empty:  # No message got in time @UndefinedVariable
                 continue
 
@@ -230,8 +285,25 @@ class ClientIPC(threading.Thread):
 
         return None
 
+    def sendRequestMessage(self, msg, data=None):
+        if data is None:
+            data = b''
+
+        if isinstance(data, six.text_type):  # Convert to bytes if necessary
+            data = data.encode('utf-8')
+
+        l = len(data)
+        msg = six.int2byte(msg) + six.int2byte(l & 0xFF) + six.int2byte(l >> 8) + data
+        self.clientSocket.sendall(msg)
+
     def requestInformation(self):
-        self.clientSocket.sendall(chr(REQ_INFORMATION))
+        self.sendRequestMessage(REQ_INFORMATION)
+
+    def sendLogin(self, username):
+        self.sendRequestMessage(REQ_LOGIN, username)
+
+    def sendLogout(self, username):
+        self.sendRequestMessage(REQ_LOGOUT, username)
 
     def messageReceived(self):
         '''

actors/src/udsactor/linux/UDSActorService.py

@@ -0,0 +1,188 @@
+# -*- 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: Adolfo Gómez, dkmaster at dkmon dot com
+'''
+from __future__ import unicode_literals
+
+from udsactor import store
+from udsactor import REST
+from udsactor import operations
+from udsactor import httpserver
+from udsactor import ipc
+
+from udsactor.log import logger
+
+from .daemon import Daemon
+from . import renamer
+
+import socket
+import time
+
+
+IPC_PORT = 39188
+
+cfg = None
+
+
+class UDSActorSvc(Daemon):
+    def __init__(self, args):
+        Daemon.__init__(self, '/var/run/udsa.pid')
+        self.isAlive = True
+        socket.setdefaulttimeout(20)
+        self.api = None
+        self.ipc = None
+        self.httpServer = None
+        self.rebootRequested = False
+        self.knownIps = []
+
+    def setReady(self):
+        self.api.setReady([(v.mac, v.ip) for v in operations.getNetworkInfo()])
+
+    def rename(self, name, user=None, oldPassword=None, newPassword=None):
+        '''
+        Renames the computer, and optionally sets a password for an user
+        before this
+        '''
+
+        # Check for password change request for an user
+        if user is not None:
+            logger.info('Setting password for user {}'.format(user))
+            try:
+                operations.changeUserPassword(user, oldPassword, newPassword)
+            except Exception as e:
+                # We stop here without even renaming computer, because the
+                # process has failed
+                raise Exception(
+                    'Could not change password for user {} (maybe invalid current password is configured at broker): {} '.format(user, unicode(e)))
+
+        renamer.rename(name)
+        self.setReady()
+
+    def interactWithBroker(self):
+        '''
+        Returns True to continue to main loop, false to stop & exit service
+        '''
+        # If no configuration is found, stop service
+        if cfg is None:
+            logger.fatal('No configuration found, stopping service')
+            return False
+
+        self.api = REST.Api(cfg['host'], cfg['masterKey'], cfg['ssl'], scrambledResponses=True)
+
+        # Wait for Broker to be ready
+        counter = 0
+        while self.isAlive:
+            try:
+                # getNetworkInfo is a generator function
+                netInfo = tuple(operations.getNetworkInfo())
+                self.knownIps = dict(((i.mac, i.ip) for i in netInfo))
+                ids = ','.join([i.mac for i in netInfo])
+                if ids == '':
+                    # Wait for any network interface to be ready
+                    logger.debug('No network interfaces found, retrying in a while...')
+                    raise Exception()
+                logger.debug('Ids: {}'.format(ids))
+                self.api.init(ids)
+                # Set remote logger to notify log info to broker
+                logger.setRemoteLogger(self.api)
+
+                break
+            except REST.InvalidKeyError:
+                logger.fatal('Can\'t sync with broker: Invalid broker Master Key')
+                return False
+            except REST.UnmanagedHostError:
+                # Maybe interface that is registered with broker is not enabled already?
+                # Right now, we thing that the interface connected to broker is
+                # the interface that broker will know, let's see how this works
+                logger.fatal('This host is not managed by UDS Broker (ids: {})'.format(ids))
+                return False
+            except Exception as e:
+                logger.debug('Exception caught: {}, retrying'.format(e.message.decode('windows-1250', 'ignore')))
+                # Any other error is expectable and recoverable, so let's wait a bit and retry again
+                # but, if too many errors, will log it (one every minute, for
+                # example)
+                counter += 1
+                if counter % 60 == 0:  # Every 5 minutes, raise a log
+                    logger.info('Trying to inititialize connection with broker (last error: {})'.format(e.message.decode('windows-1250', 'ignore')))
+                # Wait a bit before next check
+                time.sleep(1)
+
+        # Broker connection is initialized, now get information about what to
+        # do
+        counter = 0
+        while self.isAlive:
+            try:
+                logger.debug('Requesting information of what to do now')
+                info = self.api.information()
+                data = info.split('\r')
+                if len(data) != 2:
+                    logger.error('The format of the information message is not correct (got {})'.format(info))
+                    raise Exception
+                params = data[1].split('\t')
+                if data[0] == 'rename':
+                    try:
+                        if len(params) == 1:  # Simple rename
+                            self.rename(params[0])
+                        # Rename with change password for an user
+                        elif len(params) == 4:
+                            self.rename(params[0], params[1], params[2], params[3])
+                        else:
+                            logger.error('Got invalid parameter for rename operation: {}'.format(params))
+                            return False
+                        break
+                    except Exception as e:
+                        logger.error('Error at computer renaming stage: {}'.format(e.message))
+                        return False
+                else:
+                    logger.error('Unrecognized action sent from broker: {}'.format(data[0]))
+                    return False  # Stop running service
+            except REST.UserServiceNotFoundError:
+                logger.error('The host has lost the sync state with broker! (host uuid changed?)')
+                return False
+            except Exception:
+                counter += 1
+                if counter % 60 == 0:
+                    logger.warn('Too many retries in progress, though still trying (last error: {})'.format(e))
+                # Any other error is expectable and recoverable, so let's wait
+                # a bit and retry again
+                # Wait a bit before next check
+                time.sleep(1)
+
+        if self.rebootRequested:
+            try:
+                operations.reboot()
+            except Exception as e:
+                logger.error('Exception on reboot: {}'.format(e.message))
+            return False  # Stops service
+
+        return True
+
+    def run(self):
+        pass

actors/src/udsactor/linux/daemon.py

@@ -64,7 +64,7 @@ class Daemon:
                 # exit first parent
                 sys.exit(0)
         except OSError as e:
-            sys.stderr.write("fork #1 failed: %d (%s)\n" % (e.errno, e.strerror))
+            sys.stderr.write("fork #1 failed: {} ({})\n".format(e.errno, e.strerror))
             sys.exit(1)
 
         # decouple from parent environment
@@ -96,7 +96,7 @@ class Daemon:
         atexit.register(self.delpid)
         pid = str(os.getpid())
         with open(self.pidfile, 'w+') as f:
-            f.write("%s\n" % pid)
+            f.write("{}\n".format(pid))
 
     def delpid(self):
         os.remove(self.pidfile)

actors/src/udsactor/linux/renamer/debian.py

@@ -38,33 +38,28 @@ import os
 
 
 def rename(newName):
-    # If new name has "'\t'
+    '''
-    if '\t' in newName:
+    Debian renamer
-        newName, account, password = newName.split('\t')
+    Expects new host name on newName
-    else:
+    Host does not needs to be rebooted after renaming
-        account = password = None
+    '''
+    logger.debug('using Debian renamer')
 
-    logger.debug('Debian renamer')
+    with open('/etc/hostname', 'w') as hostname:
+        hostname.write(newName)
 
-    if account is not None:
+    # Force system new name
-        os.system('echo "{1}\n{1}" | /usr/bin/passwd {0} 2> /dev/null'.format(account, password))
-
-    f = open('/etc/hostname', 'w')
-    f.write(newName)
-    f.close()
     os.system('/bin/hostname %s' % newName)
 
     # add name to "hosts"
-    f = open('/etc/hosts', 'r')
+    with open('/etc/hosts', 'r') as hosts:
-    lines = f.readlines()
+        lines = hosts.readlines()
-    f.close()
+    with open('/etc/hosts', 'w') as hosts:
-    f = open('/etc/hosts', 'w')
+        hosts.write("127.0.1.1\t%s\n" % newName)
-    f.write("127.0.1.1\t%s\n" % newName)
         for l in lines:
-        if l[:9] == '127.0.1.1':
+            if l[:9] == '127.0.1.1':  # Skips existing 127.0.1.1. if it already exists
                 continue
-        f.write(l)
+            hosts.write(l)
-    f.close()
 
     return True
 

actors/src/udsactor/windows/UDSActorService.py

@@ -48,9 +48,16 @@ from udsactor import operations
 from udsactor import httpserver
 from udsactor import ipc
 
-from udsactor.windows.SENS import *  # @UnusedWildImport
 from udsactor.log import logger
 
+from .SENS import SensLogon
+from .SENS import logevent
+from .SENS import SENSGUID_EVENTCLASS_LOGON
+from .SENS import SENSGUID_PUBLISHER
+from .SENS import PROGID_EventSubscription
+from .SENS import PROGID_EventSystem
+
+
 IPC_PORT = 39188
 
 cfg = None
@@ -354,7 +361,7 @@ class UDSActorSvc(win32serviceutil.ServiceFramework):
         # ********************************
         # * Registers SENS subscriptions *
         # ********************************
-        logevent('Registring ISensLogon')
+        logevent('Registering ISensLogon')
         subscription_guid = '{41099152-498E-11E4-8FD3-10FEED05884B}'
         sl = SensLogon(self.api)
         subscription_interface = pythoncom.WrapObject(sl)