Add snippet Execute,Better Task Results , fix some bugs

migrations/017_task_results.py

@@ -1,12 +1,12 @@
-# 027_permissions.py
+# 017_permissions.py
 
 def migrate(migrator, database, fake=False, **kwargs):
-
-
     migrator.sql("""CREATE TABLE task_results(
         id serial PRIMARY KEY NOT NULL,
         task_type text,
         result text,
+        info text,
+        external_id int,
         created  timestamp not null default CURRENT_TIMESTAMP
     )""")
 

py/api/api_scanner.py

@@ -31,7 +31,7 @@ def scan_network():
     if not status:
         if start and end and port:
             db_syslog.add_syslog_event(get_myself(), "Scanner","start", get_ip(),get_agent(),json.dumps(input))
-            bgtasks.scan_with_ip(start=start,end=end,port=port,password=password,username=username)
+            bgtasks.scan_with_ip(start=start,end=end,port=port,password=password,username=username,user=get_myself())
             return buildResponse({'status': True},200)
         else:
             return buildResponse({'status': status},200)

py/api/api_snippet.py

@@ -5,15 +5,16 @@
 # MikroWizard.com , Mikrotik router management solution
 # Author: sepehr.ha@gmail.com
 
-from flask import request
+from flask import request,session
 
-from libs.db import db_user_tasks,db_syslog
+from libs.db import db_user_tasks,db_syslog,db_tasks,db_sysconfig
 from libs.webutil import app, login_required,buildResponse,get_myself,get_ip,get_agent
 from functools import reduce
+import bgtasks
 import operator
 import logging
 import json
-
+import datetime
 log = logging.getLogger("api.snippet")
 
 @app.route('/api/snippet/list', methods = ['POST'])
@@ -95,4 +96,90 @@ def user_snippet_delete():
         snippet=db_user_tasks.delete_snippet(id)
         return buildResponse({"result":"success"}, 200)
     else:
-        return buildResponse({"result":"failed","err":"Failed to delete snippet"}, 200)
+        return buildResponse({"result":"failed","err":"Failed to delete snippet"}, 200)
+    
+@app.route('/api/snippet/exec', methods = ['POST'])
+@login_required(role='admin',perm={'task':'write'})
+def exec_snippet():
+    """crate user task"""
+    input = request.json
+    description=input.get('description',None)
+    snippetid=input.get('id',False)
+    members=input.get('members', False)
+    task_type=input.get('task_type',"backup")
+    selection_type=input.get('selection_type',False)
+    # taskdata=input.get('data',False)
+    utasks=db_user_tasks.UserTasks
+    
+    # todo
+    # add owner check devids and dev groups with owner
+    if not description:
+        return buildResponse({'status': 'failed'},200,error="Wrong name/desc")
+    #check if cron is valid and correct
+    taskdata={}
+    taskdata['memebrs']=members
+    taskdata['owner']=members
+    snipet=db_user_tasks.get_snippet(snippetid)
+    if snipet:
+        taskdata['snippet']={'id':snipet.id,'code':snipet.content,'description':snipet.description,'name':snipet.name}
+    else:
+        return buildResponse({'status': 'failed'}, 200, error="Wrong snippet")
+
+    if selection_type not in ["devices","groups"]:
+        return buildResponse({'status': 'failed'}, 200, error="Wrong member type")
+    if task_type != 'snipet_exec':
+        return buildResponse({'status': 'failed'}, 200, error="Wrong task type")
+    try:
+        data={
+            'name':snipet.name,
+            'description':description,
+            'snippetid':int(snippetid),
+            'cron':None,
+            'desc_cron': None,
+            'action': 'snipet_exec',
+            'task_type':'snipet_exec',
+            'selection_type':selection_type,
+            'data':json.dumps(taskdata),
+            'created': datetime.datetime.now()
+        }
+        task=utasks.create(**data)
+        status=db_tasks.exec_snipet_status().status
+        uid = session.get("userid") or False
+        default_ip=db_sysconfig.get_sysconfig('default_ip')
+        if not uid:
+            return buildResponse({'result':'failed','err':"No User"}, 200)
+        if not status:
+            bgtasks.exec_snipet(task=task,default_ip=default_ip,devices=members,uid=uid)
+            res={'status': True}
+        else:
+            res={'status': status}
+        #add members to task
+        db_syslog.add_syslog_event(get_myself(), "Snippet","execute", get_ip(),get_agent(),json.dumps(input))
+        return buildResponse([{'status': 'success'}],200)
+    except Exception as e:
+        log.error(e)
+        return buildResponse({'status': 'failed','massage':str(e)},200)    
+
+    
+@app.route('/api/snippet/executed', methods = ['POST'])
+@login_required(role='admin',perm={'task':'write'})
+def get_executed_snippet():
+    """crate user task"""
+    input = request.json
+    id=input.get('id', False)
+    snipet=db_user_tasks.get_snippet(id)
+    if not snipet:
+        return buildResponse({'status': 'failed'}, 200, error="Wrong snippet")
+    utasks=db_user_tasks.UserTasks
+    tasks=utasks.select().where(utasks.snippetid==id).where(utasks.task_type=='snipet_exec')
+    taks_ids=[task.id for task in tasks]
+    task_res=db_tasks.TaskResults
+    executed_tasks=task_res.select().where(task_res.external_id<<taks_ids).order_by(task_res.id.desc())
+    executed_tasks=list(executed_tasks.dicts())
+    for task in executed_tasks:
+        task['result']=json.loads(task['result'])
+        task['info']=json.loads(task['info'])
+    if executed_tasks:
+        return buildResponse(executed_tasks, 200)
+    else:
+        return buildResponse({'status': 'failed'}, 200)

py/api/api_user_tasks.py

@@ -46,6 +46,7 @@ def user_tasks_list():
         clauses.append(utaks.task_type == task_type)
     if not ISPRO:
         clauses.append(utaks.task_type != 'firmware')
+    clauses.append(utaks.task_type != 'snipet_exec')
     expr=""
     logs = []
     selector=[utaks.id,utaks.name,utaks.description,utaks.desc_cron,utaks.action,utaks.task_type,utaks.dev_ids,utaks.snippetid,utaks.data,utaks.cron,utaks.selection_type,utaks.created]
@@ -108,19 +109,18 @@ def user_tasks_create():
         if len(members):
             db_user_tasks.add_member_to_task(task.id, members, selection_type)
         taskid=task.id
-        if task_type=="backup":
+        crontab = CronTab(user=True)
-            crontab = CronTab(user=True)
+        directory=Path(app.root_path).parent.absolute()
-            directory=Path(app.root_path).parent.absolute()
+        command = "python3 {}/task_run.py {}".format(directory,taskid)
-            command = "python3 {}/task_run.py {}".format(directory,taskid)
+        comment = "MikroWizard task #" + "taskid:{};".format(taskid)
-            comment = "MikroWizard task #" + "taskid:{};".format(taskid)
+        jobs = crontab.find_comment(comment)
+        if len(list(jobs)) > 0:
             jobs = crontab.find_comment(comment)
-            if len(list(jobs)) > 0:
+            crontab.remove(jobs)
-                jobs = crontab.find_comment(comment)
-                crontab.remove(jobs)
-                crontab.write()
-            job = crontab.new(command=command,comment=comment)
-            job.setall(cron)
             crontab.write()
+        job = crontab.new(command=command,comment=comment)
+        job.setall(cron)
+        crontab.write()
         db_syslog.add_syslog_event(get_myself(), "Task","Create", get_ip(),get_agent(),json.dumps(input))
         return buildResponse([{'status': 'success',"taskid":taskid}],200)
     except Exception as e:

py/bgtasks.py

@@ -6,7 +6,7 @@
 # Author: sepehr.ha@gmail.com thanks to Tomi.Mickelsson@iki.fi
 
 from uwsgidecorators import spool
-
+from playhouse.shortcuts import model_to_dict
 from libs import util
 import time
 from libs.db import db_tasks,db_device,db_events,db_user_group_perm,db_device
@@ -20,6 +20,7 @@ import socket
 from libs.check_routeros.routeros_check.resource import RouterOSCheckResource
 from typing import Dict
 import json
+import datetime
 
 sensor_pile = queue.LifoQueue()
 other_sensor_pile = queue.LifoQueue()
@@ -27,7 +28,9 @@ other_sensor_pile = queue.LifoQueue()
 import logging
 log = logging.getLogger("bgtasks")
 
-
+def serialize_datetime(obj): 
+    if isinstance(obj, datetime.datetime): 
+        return obj.isoformat() 
 
 @spool(pass_arguments=True)
 def check_devices_for_update(*args, **kwargs):
@@ -267,6 +270,11 @@ def scan_with_ip(*args, **kwargs):
         mikrotiks=[]
         scan_results=[]
         dev_number=0
+        info={
+            'user':kwargs.get('user','Unknown'),
+            'start_ip':start_ip,
+            'end_ip':end_ip
+        }
         for ip_int in range(int(start_ip), int(end_ip)):
             ip=str(ipaddress.IPv4Address(ip_int))
             sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
@@ -358,7 +366,7 @@ def scan_with_ip(*args, **kwargs):
                     continue
             sock.close()
         try:
-            db_tasks.add_task_result('ip-scan', json.dumps(scan_results))
+            db_tasks.add_task_result('ip-scan', json.dumps(scan_results),json.dumps(info))
         except:
             pass
         #ugly hack to reset sequnce number if device id
@@ -381,4 +389,63 @@ def scan_with_ip(*args, **kwargs):
         log.error(e)
         task.status=0
         task.save()
-        return True
+        return True
+    
+    
+    
+    
+    
+@spool(pass_arguments=True)
+def exec_snipet(*args, **kwargs):
+    task=db_tasks.exec_snipet_status()
+    if not task.status:
+        task.status=1
+        task.save()
+        default_ip=kwargs.get('default_ip',False)
+        try:
+            if kwargs.get('devices',False) and kwargs.get('task',False):
+                devids=kwargs.get('devices',False)
+                devs=False
+                uid=kwargs.get('uid',False)
+                utask=kwargs.get('task',False)
+                taskdata=json.loads(utask.data)
+                if "0" == devids:
+                    devs=list(db_user_group_perm.DevUserGroupPermRel.get_user_devices(uid))
+                else:
+                    devids=devids
+                    devs=list(db_user_group_perm.DevUserGroupPermRel.get_user_devices_by_ids(uid,devids))
+                num_threads = len(devs)
+                q = queue.Queue()
+                threads = []
+                log.error(devs)
+                for dev in devs:
+                    peer_ip=dev.peer_ip if dev.peer_ip else default_ip
+                    if not peer_ip and '[mikrowizard]' in taskdata['snippet']['code']:
+                        log.error("no peer ip")
+                        num_threads=num_threads-1
+                        continue
+                    snipet_code=taskdata['snippet']['code']
+                    if '[mikrowizard]' in taskdata['snippet']['code']:
+                        snipet_code=snipet_code.replace('[mikrowizard]', peer_ip)
+                    t = Thread(target=util.run_snippets, args=(dev, snipet_code, q))
+                    t.start()
+                    threads.append(t)
+                for t in threads:
+                    t.join()
+                res=[]
+                for _ in range(num_threads):
+                    qres=q.get()
+                    res.append(qres)
+                try:
+                    db_tasks.add_task_result('snipet_exec', json.dumps(res),json.dumps(model_to_dict(utask),default=serialize_datetime),utask.id)
+                except Exception as e:
+                    log.error(e)
+                    pass
+        except Exception as e:
+            log.error(e)
+            task.status=0
+            task.save()
+            return False
+    task.status=0
+    task.save()
+    return False

py/libs/db/db_tasks.py

@@ -48,10 +48,15 @@ def downloader_job_status():
 def firmware_service_status():
     return (Tasks.select().where(Tasks.signal == 150).get())
 
+def exec_snipet_status():
+    return (Tasks.select().where(Tasks.signal == 160).get())
+
 
 class TaskResults(BaseModel):
     task_type = TextField()
     result = DateTimeField()
+    info = TextField()
+    external_id = IntegerField()
     created = DateTimeField()
 
     class Meta:
@@ -60,8 +65,8 @@ class TaskResults(BaseModel):
         # whether the index is unique or not.
         db_table = 'task_results'
 
-def add_task_result(task_type,result):
+def add_task_result(task_type,result,info=None,eid=None):
-    tr = TaskResults(task_type=task_type, result=result)
+    tr = TaskResults(task_type=task_type, result=result,info=info,external_id=eid)
     tr.save()
 
 # --------------------------------------------------------------------------

py/libs/util.py

@@ -599,7 +599,7 @@ def check_update(options,router=False):
         return False,False,False,False
 
 def log_alert(type,dev,massage):
-    log.error("Alert: {} {}".format(type,massage))
+    log.error("Alert: {} {} Device: {} ".format(type,massage,dev.ip))
 
 def backup_routers(dev,q):
     status=backup_router(dev)
@@ -607,7 +607,7 @@ def backup_routers(dev,q):
 
 def run_snippets(dev, snippet,q):
     result=run_snippet(dev, snippet)
-    q.put({"id": dev.id, "result":result})
+    q.put({"devid": dev.id,"devip": dev.ip,"devname": dev.name, "status":True if result else False , "result":result if result else 'Exec Failed'})
     return result
 
 def run_snippet(dev, snippet):
@@ -638,7 +638,8 @@ def run_snippet(dev, snippet):
             try:
                 ssh=SSH_Helper(options)
                 result=ssh.exec_command(snippet)
-                
+                if not result:
+                    result="executed successfully"
             except Exception as e:
                 log.error(e)
                 log_alert('ssh',dev,'During backup ssh error')
@@ -895,9 +896,6 @@ def download_firmware_to_repository(version,q,arch="all",all_package=False):
         log.error(links)
         firm=db_firmware.Firmware()
         for lnk in links:
-            log.error("oooooooooooooooooooooooooooooooooooooooooo")
-            log.error(lnk)
-            log.error("oooooooooooooooooooooooooooooooooooooooooo")
             if all_package and arch+"-allpackage" == lnk:
                 arch_togo=lnk
                 link=links[lnk]["link"]