Changeset 3547

Timestamp:

Feb 7, 2017, 2:44:17 PM (4 years ago)

Author:

hectorgh

Message:

VariablesManager? changes. Read changelog. startup thread is launched after service is up

Location:

n4d/trunk/fuentes

Files:

• debian/changelog (modified) (1 diff)
• install-files/etc/n4d/conf.d/VariablesManager (modified) (1 diff)
• install-files/usr/share/n4d/python-plugins/VariablesManager.py (modified) (18 diffs)
• install-files/usr/share/n4d/xmlrpc-server/core.py (modified) (6 diffs)
• install-files/usr/share/n4d/xmlrpc-server/server.py (modified) (2 diffs)

n4d/trunk/fuentes/debian/changelog

@@ +1 @@
+n4d (0.107) xenial; urgency=high
+
+  * Startups, one-shots and cron threads are executed after service is up
+  * VariablesManager now supports server-client communication. If a variable
+    is modified in the server, clients are notified of such changes.
+  * VariablesManager as a Variables Server keeps track of a list of 
+    alive clients using their IP and MACs as identifiers.
+  * VariablesManager now supports triggers executed when a certain variable is
+    modified.
+  * set_variable makes the distinction between setting a new value and 
+    trying to set a value that is already stored in the variable.
+
+ -- Hector Garcia Huerta <hectorgh@gmail.com>  Tue, 07 Feb 2017 14:36:19 +0100
+
 n4d (0.106) xenial; urgency=high
 

n4d/trunk/fuentes/install-files/etc/n4d/conf.d/VariablesManager

@@ -15 +15 @@
 backup=adm,admins,admin
 restore=adm,admins,admin
+(ip)register_instance=anonymous
+(ip)server_changed=anonymous

n4d/trunk/fuentes/install-files/usr/share/n4d/python-plugins/VariablesManager.py

@@ -11 +11 @@
 import tarfile
 import threading
+import subprocess
+import time
+import string
 
 class VariablesManager:
@@ -24 +27 @@
         def __init__(self):
                 
+                self.instance_id="".join(random.sample(string.letters+string.digits, 50))
+                self.server_instance_id=None
                 self.variables={}
                 self.variables_ok=False
+                self.variables_clients={}
+                self.variables_triggers={}
+                t=threading.Thread(target=self.check_clients,args=())
+                t.daemon=True
+                t.start()
+                
                 if os.path.exists(VariablesManager.LOCK_FILE):
                         os.remove(VariablesManager.LOCK_FILE)
@@ -59 +70 @@
         #def __init__
         
-        # DO NOT TOUCH THIS
         
         def startup(self,options):
-                pass
-                
-        # DONE ============
+
+                if "REMOTE_VARIABLES_SERVER" in self.variables:
+                        self.register_n4d_instance_to_server()
+                        
+        #def startup
+        
+        
+        def is_ip_in_range(self,ip,network):
+                
+                try:
+                        return netaddr.ip.IPAddress(ip) in netaddr.IPNetwork(network).iter_hosts()
+                except:
+                        return False
+                        
+        #def is_ip_in_range
+        
+
+        def get_net_size(self,netmask):
+                
+                netmask=netmask.split(".")
+                binary_str=""
+                for octet in netmask:
+                        binary_str += bin(int(octet))[2:].zfill(8)
+                        
+                return str(len(binary_str.rstrip('0')))
+                
+        #def get_net_size
+
+
+        def get_ip(self):
+                
+                for item in netifaces.interfaces():
+                        tmp=netifaces.ifaddresses(item)
+                        if tmp.has_key(netifaces.AF_INET):
+                                if tmp[netifaces.AF_INET][0].has_key("broadcast") and tmp[netifaces.AF_INET][0]["broadcast"]=="10.0.2.255":
+                                        return tmp[netifaces.AF_INET][0]["addr"]
+                return None
+                
+        #def get_ip
+        
+
+        def route_get_ip(self,ip):
+                
+                p=subprocess.Popen(["ip route get %s"%ip],shell=True,stdout=subprocess.PIPE,stderr=subprocess.PIPE).communicate()
+                if "dev" in p[0]:
+                        dev=p[0].split("dev ")[1].split(" ")[0]
+                else:
+                        dev=None
+                return dev
+                
+        #def route_get_ip
+                
+
+        def get_mac_from_device(self,dev):
+
+                for item in netifaces.interfaces():
+                        
+                        try:
+                                i=netifaces.ifaddresses(item)
+                                mac=i[17][0]["addr"]
+                                broadcast=i[2][0]["broadcast"]
+                                network=broadcast
+                                netmask=i[2][0]["netmask"]
+                                network+="/%s"%self.get_net_size(netmask)
+                                ip=i[2][0]["addr"]
+                        except Exception as e:
+                                continue
+                        
+                        if dev=="lo":
+                                return mac
+                        
+                        if item==dev:
+                                return mac
+                                
+                return None
+
+        #def get_mac_from_device_in_server_network
+        
+        
+        def register_instance(self,autocompleted_secured_ip,mac):
+
+                client={}
+                client["last_check"]=int(time.time())
+                client["missed_pings"]=0
+                client["ip"]=autocompleted_secured_ip
+                self.variables_clients[mac]=client
+                
+                return self.instance_id
+
+        #def register_instance
+        
+
+        def register_n4d_instance_to_server(self):
+                
+                try:
+                        server_ip=socket.gethostbyname(self.variables["REMOTE_VARIABLES_SERVER"][u"value"])
+                        if self.get_ip()!=server_ip:
+                        
+                                c=xmlrpclib.ServerProxy("https://%s:9779"%server_ip)
+                                mac=self.get_mac_from_device(self.route_get_ip(server_ip))
+                                self.server_instance_id=c.register_instance("","VariablesManager","",mac)
+                        
+                except Exception as e:
+
+                        return None
+
+        #def register_n4d_instance_to_server    
+        
+        
+        def check_clients(self):
+                
+                while True:
+                        
+                        for item in self.variables_clients:
+                                ip=self.variables_clients[item]["ip"]
+                                sys.stdout.write("[VariablesManager] Checking client { MAC:%s IP:%s } ... "%(item,ip))
+                                c=xmlrpclib.ServerProxy("https://%s:9779"%ip)
+                                try:
+                                        c.get_methods()
+                                        self.variables_clients[item]["last_check"]=time.time()
+                                        self.variables_clients[item]["missed_pings"]=0
+                                        print("OK")
+                                except:
+                                        self.variables_clients[item]["missed_pings"]+=1
+                                        print("FAILED")
+                                        if self.variables_clients[item]["missed_pings"] >=3:
+                                                print "[VariablesManager] Removing client due to too many missed pings."
+                                                self.variables_clients.pop(item)
+                        
+                        time.sleep(60*5)
+                
+                
+        #def check_clients
+        
+        def notify_changes(self,variable):
+                
+                print "[VariablesManager] Notifying changes... "
+                
+                for mac in self.variables_clients:
+                        
+                        ip=self.variables_clients[mac]["ip"]
+                        c=xmlrpclib.ServerProxy("https://%s:9779"%ip)
+                        try:
+                                c.server_changed("","VariablesManager","",self.instance_id,variable)
+                                
+                        except:
+                                self.variables_clients[mac]["missed_pings"]+=1
+                                
+                        self.variables_clients[mac]["last_check"]=time.time()
+                
+        #def announce_changes
+        
+        
+        def server_changed(self,autocompleted_server_ip,server_instance_id,variable_name):
+
+                if server_instance_id==self.server_instance_id:
+
+                        if self.variables["REMOTE_VARIABLES_SERVER"] [u'value']!=autocompleted_server_ip:
+                                return False
+
+                        print "[VariablesManager] Server instance ID validated"
+
+                        t=threading.Thread(target=self.execute_trigger,args=(variable_name,))
+                        t.daemon=True
+                        t.start()
+                        
+                        return True
+                        
+                else:
+                        
+                        return False
+                
+        #def server_changed
+        
+        
+        def execute_trigger(self,variable_name):
+                
+                if variable_name in self.variables_triggers:
+                        for i in self.variables_triggers[variable_name]:
+                                class_name,function=i
+                                try:
+                                        print "[VariablesManager] Executing %s.%s trigger..."%(class_name,function.im_func.func_name)
+                                        function()
+                                except Exception as e:
+                                        print e
+                                        pass
+                
+        #def execute_trigger
+        
+        
+        def register_trigger(self,variable_name,class_name,function):
+                
+                if variable_name not in self.variables_triggers:
+                        self.variables_triggers[variable_name]=[]
+                        
+                self.variables_triggers[variable_name].append((class_name,function))
+                
+        #def register_trigger
         
         
@@ -86 +291 @@
                 
         #def backup
+
         
         def restore(self,file_path=None):
@@ -166 +372 @@
                 ret=re.findall(pattern,value)
                 
-                print "hola",ret
-                
                 for item in ret:
                         tmp=item.replace("_@START@_","")
@@ -178 +382 @@
                 return value
                 
-                
-                
         #def remove_calculated_chars
         
@@ -188 +390 @@
                         if not self.variables[item].has_key("volatile"):
                                 self.variables[item]["volatile"]=False
-                        
-                
                 
         #def add_volatile_info
@@ -215 +415 @@
                                                 
                 return ret.strip("\n")
-                                        
-                
                 
         #def  showvars
+
         
         def get_variables(self):
@@ -314 +513 @@
                 return [True,""]
                                 
-                
-                        
-                        
-                '''
-                
-                if os.path.exists(VariablesManager.INBOX):
-                        
-                        for file in os.listdir(VariablesManager.INBOX):
-                                file_path=VariablesManager.INBOX+file
-                                
-                                try:
-                                        execfile(file_path)
-                                        os.remove(file_path)
-                                except Exception as e:
-                                        self.log(file_path + ": " + str(e))
-                        
-                '''
-                
         #def read_inbox
+
         
         def empty_trash(self,force_write=False):
@@ -380 +562 @@
         #def empty_trash
         
-        def get_ip(self):
-                
-                for item in netifaces.interfaces():
-                        tmp=netifaces.ifaddresses(item)
-                        if tmp.has_key(netifaces.AF_INET):
-                                if tmp[netifaces.AF_INET][0].has_key("broadcast") and tmp[netifaces.AF_INET][0]["broadcast"]=="10.0.2.255":
-                                        return tmp[netifaces.AF_INET][0]["addr"]
-                return None
-                
-        #def get_ip
 
         def get_variable_list(self,variable_list,store=False,full_info=False):
@@ -415 +587 @@
                         return None
                         
-                
-        
                 if name in self.variables and self.variables[name].has_key("function"):
                         try:
@@ -487 +657 @@
 
                 if name in self.variables:
+                        
+                        if value == self.variables[name][u"value"]:
+                                return [True,"Variable already contained that value"]
+                        
                         if type(value)==type(""):
                                 self.variables[name][u"value"]=unicode(value).encode("utf-8")
@@ -505 +679 @@
                                 if "description" not in self.variables["name"]:
                                         self.variables[name]["description"]=""
-                                        
+                        
+                        t=threading.Thread(target=self.notify_changes,args=(name,))
+                        t.daemon=True
+                        t.start()
+                        
                         return [True,""]
                 else:
@@ -512 +690 @@
                 
         #def set_variable
+
         
         def add_variable(self,name,value,function,description,depends,volatile=False,root_protected=False):
@@ -537 +716 @@
                 else:
                         return [False,"Variable already exists. Use set_variable"]
-                
+                        
+        #def add_variable
+
+
         def write_file(self,fname=None):
-                
-                '''
-                
-                try:
-                        while os.path.exists(VariablesManager.LOCK_FILE):
-                                time.sleep(2)
-                        f=open(VariablesManager.LOCK_FILE,"w")
-                        f.close()
-                        tmp={}
-                        for item in self.variables:
-                                if self.variables[item].has_key("volatile") and self.variables[item]["volatile"]==False:
-                                        tmp[item]=self.variables[item]
-
-                        if fname==None:
-                                f=open(VariablesManager.VARIABLES_FILE,"w")
-                        else:
-                                f=open(fname,"w")
-                                
-                        data=unicode(json.dumps(tmp,indent=4,encoding="utf-8",ensure_ascii=False)).encode("utf-8")
-                        f.write(data)
-                        f.close()
-                        os.remove(VariablesManager.LOCK_FILE)
-                        return True
-                        
-                except Exception as e:
-                        os.remove(VariablesManager.LOCK_FILE)
-                        print(e)
-                        return False
-                        
-                '''
                 
                 try:
@@ -604 +756 @@
                         return False
                 
-                        
-                
         #def write_file
+
 
         def chmod(self,file,mode):
@@ -618 +769 @@
                         os.umask(prevmask)
                         return False
+                        
         #def chmod
+        
         
         def init_variable(self,variable,args={},force=False,full_info=False):
@@ -638 +791 @@
                         return (False,e)
                 
-                
         #def init_variable
         

n4d/trunk/fuentes/install-files/usr/share/n4d/xmlrpc-server/core.py

@@ -274 +274 @@
                         
 
-t=threading.Thread(target=one_shot)
-t.daemon=True
-t.start()
+n4d_core_one_shot_t=threading.Thread(target=one_shot)
+n4d_core_one_shot_t.daemon=True
+# Will be executed in server.py after socket is ready
+#n4d_core_one_shot_t.start()
 
 
@@ -497 +498 @@
                                 count+=1
 
-                        p=threading.Thread(target=startup_launcher,args=(objects,))
-                        p.daemon=True
-                        p.start()
+                        load_t=threading.Thread(target=startup_launcher,args=(objects,))
+                        load_t.daemon=True
+                        load_t.start()
                         #startup_launcher(objects)
                         
@@ -646 +647 @@
 
 
-p=threading.Thread(target=startup_launcher,args=(objects,))
-p.daemon=True
-p.start()
+n4d_core_startup_t=threading.Thread(target=startup_launcher,args=(objects,))
+n4d_core_startup_t.daemon=True
+# Will be executed in server.py after socket is ready
+#n4d_core_startup_t.start()
 
 
@@ -705 +707 @@
 n4d_cron_thread=threading.Thread(target=n4d_cron)
 n4d_cron_thread.daemon=True
-n4d_cron_thread.start()
+# Will be executed in server.py after socket is ready
+#n4d_cron_thread.start()
         
 # OLD FUNCTION LIST
@@ -916 +919 @@
 #def validate_user
 
+def exec_threads():
+        
+        global n4d_core_startup_t, n4d_core_one_shot_t,n4d_cron_thread
+
+        n4d_core_one_shot_t.start()
+        n4d_core_startup_t.start()
+        n4d_cron_thread.start()
+        
+#def exec_threads
+
 
 srv_logger.close()
@@ -947 +960 @@
                 self.run=True
                 import threading
-                t=threading.Thread(target=self.clear_credentials,args=())
-                t.daemon=True
-                t.start()
+                n4d_core_clear_t=threading.Thread(target=self.clear_credentials,args=())
+                n4d_core_clear_t.daemon=True
+                n4d_core_clear_t.start()
                 
         def launch_triggers(self):

n4d/trunk/fuentes/install-files/usr/share/n4d/xmlrpc-server/server.py

@@ -325 +325 @@
 
 
-        def startup(self):
+        def startup(self,funct):
                 'run until quit signaled from keyboard...'
                 print '[SERVER] Starting; Press CTRL-C to quit ...'
+                funct()
                 while True:
                         try:
@@ -393 +394 @@
         l=obj.Core()
         server_address = ('', 9779) # (address, port)
-        server = SecureDocXMLRPCServer(l, server_address, DEFAULTKEYFILE, DEFAULTCERTFILE)      
-        #server.register_introspection_functions()
-        '''
-        for plugin in l.cm.plugins:
-                for method in plugin.function:
-                        server.register_function(getattr(l.objects[plugin.class_name],method))
-        '''
-        
+        tries=1
+        max_tries=5
+        while tries <= max_tries:
+                try:
+                        server = SecureDocXMLRPCServer(l, server_address, DEFAULTKEYFILE, DEFAULTCERTFILE)      
+                        tries=max_tries+1
+                except:
+                        print "[SERVER] Failed to obtain socket. Trying again [%s/%s]..."%(tries,max_tries)
+                        time.sleep(1)
+                        tries+=1
+                        if tries >= 5:
+                                print "[SERVER] Unable to open socket. Exiting..."
+                                sys.exit(1)
                         
         sa = server.socket.getsockname()
         print "[SERVER] HTTPS on", sa[0], "port", sa[1]
-        server.startup()
-
-
+        server.startup(obj.exec_threads)
+        
+
+