mirror of
https://github.com/MikroWizard/mikroman.git
synced 2025-06-21 03:55:39 +02:00
169 lines
6.5 KiB
Python
169 lines
6.5 KiB
Python
#!/usr/bin/python
|
|
# -*- coding: utf-8 -*-
|
|
|
|
# api_firmware.py: API for managing firmware
|
|
# MikroWizard.com , Mikrotik router management solution
|
|
# Author: sepehr.ha@gmail.com
|
|
|
|
from flask import request, jsonify,session,send_file
|
|
from playhouse.shortcuts import model_to_dict
|
|
import datetime
|
|
from libs.db import db_tasks,db_sysconfig,db_firmware,db_syslog
|
|
from libs import util,firm_lib
|
|
from libs.webutil import app, login_required, get_myself,buildResponse,get_myself,get_ip,get_agent
|
|
import bgtasks
|
|
import re
|
|
import logging
|
|
|
|
log = logging.getLogger("api.firmware")
|
|
import json
|
|
|
|
@app.route('/api/firmware/check_firmware_update', methods = ['POST'])
|
|
@login_required(role='admin',perm={'device':'write'})
|
|
def check_firmware_update():
|
|
"""Chck fimware update status"""
|
|
input = request.json
|
|
devids=input.get('devids',"0")
|
|
status=db_tasks.update_check_status().status
|
|
uid = session.get("userid") or False
|
|
if not uid:
|
|
return buildResponse({'result':'failed','err':"No User"}, 200)
|
|
#if devices is [0] then check firmware for all devices of user
|
|
if not status:
|
|
bgtasks.check_devices_for_update(devices=devids,uid=uid)
|
|
db_syslog.add_syslog_event(get_myself(), "Firmware","Check", get_ip(),get_agent(),json.dumps(input))
|
|
res={'status': True}
|
|
else:
|
|
res={'status': status}
|
|
|
|
return buildResponse(res,200)
|
|
|
|
@app.route('/api/firmware/check_task_status', methods = ['GET'])
|
|
@login_required(role='admin',perm={'device':'read'})
|
|
def check_task_status():
|
|
"""Return firmware update check service status"""
|
|
status=db_tasks.update_check_status().status
|
|
return jsonify({'status': status})
|
|
|
|
@app.route('/api/firmware/update_firmware', methods = ['POST'])
|
|
@login_required(role='admin',perm={'device':'full'})
|
|
def update_device():
|
|
"""Update devices"""
|
|
status=db_tasks.update_job_status().status
|
|
input=request.json
|
|
devids=input.get('devids',"0")
|
|
uid = session.get("userid") or False
|
|
if not uid:
|
|
return buildResponse({'result':'failed','err':"No User"}, 200)
|
|
if not status:
|
|
db_syslog.add_syslog_event(get_myself(), "Firmware","update", get_ip(),get_agent(),json.dumps(input))
|
|
bgtasks.update_device(devices=devids,uid=uid)
|
|
res={'status': True}
|
|
else:
|
|
res={'status': status}
|
|
return buildResponse(res,200)
|
|
|
|
@app.route('/api/firmware/get_firms', methods = ['POST'])
|
|
@login_required(role='admin',perm={'settings':'full'})
|
|
def get_firms():
|
|
"""get list of of downloaded firmwares in local repo"""
|
|
input = request.json or {}
|
|
page = input.get('page')
|
|
size = input.get('size')
|
|
search = input.get('search')
|
|
|
|
reply = db_firmware.query_firms(page, size, search).dicts()
|
|
data={
|
|
"firms":reply,
|
|
"updateBehavior":db_sysconfig.get_sysconfig("old_firmware_action"),
|
|
"firmwaretoinstall":db_sysconfig.get_sysconfig("latest_version"),
|
|
"firmwaretoinstallv6":db_sysconfig.get_sysconfig("old_version"),
|
|
}
|
|
return buildResponse(data, 200)
|
|
|
|
@app.route('/api/firmware/get_downloadable_firms', methods = ['POST'])
|
|
@login_required(role='admin',perm={'settings':'full'})
|
|
def get_downloadable_firms():
|
|
"""get list of availble Firmwares from Mikrotik Official webstire"""
|
|
input = request.json or {}
|
|
versions=firm_lib.get_mikrotik_versions()
|
|
versions = sorted(versions, key=lambda x: [int(y) if y.isdigit() else int(re.sub(r'\D', '', y)) for y in x.split('.')])
|
|
|
|
return buildResponse({"versions":versions}, 200)
|
|
|
|
@app.route('/api/firmware/download_firmware_to_repository', methods = ['POST'])
|
|
@login_required(role='admin',perm={'settings':'full'})
|
|
def download_firmware_to_repository():
|
|
"""Download Firmware from Mikrotik Official website"""
|
|
input = request.json or {}
|
|
version=input.get('version')
|
|
status=db_tasks.downloader_job_status().status
|
|
|
|
if not status:
|
|
db_syslog.add_syslog_event(get_myself(), "Firmware","Download", get_ip(),get_agent(),json.dumps(input))
|
|
bgtasks.download_firmware(version=version)
|
|
return buildResponse({'status': True}, 200)
|
|
else:
|
|
return buildResponse({'status': status}, 200)
|
|
|
|
@app.route('/api/firmware/delete_from_repository', methods = ['POST'])
|
|
@login_required(role='admin',perm={'settings':'full'})
|
|
def delete_from_repository():
|
|
"""Delete File from Repo"""
|
|
input = request.json or {}
|
|
id = input.get('id',False)
|
|
firmware=db_firmware.get_firm(id)
|
|
log.error(firmware)
|
|
if firmware:
|
|
#Delete file from location
|
|
location=firmware.location
|
|
res=util.delete_file(location)
|
|
log.error(res)
|
|
if res:
|
|
firmware.delete_instance()
|
|
return buildResponse({'status': True}, 200)
|
|
return buildResponse({'status': False},200,error="Firm Already deleted")
|
|
|
|
@app.route('/api/firmware/update_firmware_settings', methods = ['POST'])
|
|
@login_required(role='admin',perm={'settings':'write'})
|
|
def update_firmware_settings():
|
|
"""Change system settings for firmware update"""
|
|
input = request.json or {}
|
|
updateBehavior=input.get('updatebehavior')
|
|
firmwaretoinstall=input.get('firmwaretoinstall')
|
|
firmwaretoinstallv6=input.get('firmwaretoinstallv6')
|
|
db_sysconfig.update_sysconfig("old_firmware_action", updateBehavior)
|
|
db_sysconfig.update_sysconfig("latest_version", firmwaretoinstall)
|
|
db_sysconfig.update_sysconfig("old_version", firmwaretoinstallv6)
|
|
db_syslog.add_syslog_event(get_myself(), "Firmware","settings", get_ip(),get_agent(),json.dumps(input))
|
|
return buildResponse({'status': True}, 200)
|
|
|
|
def serialize_datetime(obj):
|
|
if isinstance(obj, datetime.datetime):
|
|
return obj.isoformat()
|
|
|
|
@app.route('/api/firmware/get_firmware/<firmid>', methods = ['POST','GET'])
|
|
def get_firmware(firmid):
|
|
"""Download firmware of given id from repo"""
|
|
firm=db_firmware.get_firm(firmid)
|
|
dev_ip=request.remote_addr
|
|
# log.error(dev_ip)
|
|
# if dev_ip:
|
|
# dev=db_device.query_device_by_ip(dev_ip)
|
|
# if not dev:
|
|
# return buildResponse({'result':'failed', 'err':"Device not found"}, 200)
|
|
|
|
if firm:
|
|
data={
|
|
"devip":dev_ip,
|
|
# "devid":dev.id,
|
|
"firm":model_to_dict(firm),
|
|
}
|
|
db_syslog.add_syslog_event(get_myself(), "Firmware","download", get_ip(),get_agent(),json.dumps(data,default=serialize_datetime))
|
|
# if dev.arch != firm.architecture:
|
|
# return buildResponse({'result':'failed','err':"Wrong architecture"}, 200)
|
|
path=firm.location
|
|
return send_file(path, as_attachment=True)
|
|
# log.error(dev)
|
|
return buildResponse({'result':'failed','err':"somthing went wrong"}, 200)
|
|
|