459 lines
18 KiB
Python
459 lines
18 KiB
Python
#!/usr/bin/python3
|
|
# -*- coding: utf-8 -*-
|
|
import os
|
|
import sys
|
|
import json
|
|
import socket
|
|
import ssl
|
|
import logging
|
|
from flask import Flask, request, jsonify, g, render_template
|
|
from contextlib import contextmanager
|
|
from dotenv import load_dotenv
|
|
from werkzeug.exceptions import HTTPException
|
|
import database
|
|
import jwt
|
|
import datetime
|
|
from functools import wraps
|
|
from werkzeug.security import generate_password_hash, check_password_hash
|
|
|
|
# Загрузка .env файла для безопасности
|
|
load_dotenv()
|
|
|
|
# Инициализация БД при запуске
|
|
database.init_db()
|
|
|
|
# --- Настройки безопасности ---
|
|
SECRET_KEY = os.environ.get('APP_SECRET_KEY', 'super-secret-key-change-me')
|
|
DEFAULT_ADMIN = os.environ.get('MT_USER', 'admin') # Используем MT_USER как дефолтный логин
|
|
DEFAULT_PASS = os.environ.get('ADMIN_PASS', 'admin123')
|
|
|
|
def ensure_admin_exists():
|
|
"""Создает дефолтного админа, если таблица пуста"""
|
|
import sqlite3
|
|
conn = database.get_db_connection()
|
|
count = conn.execute('SELECT COUNT(*) FROM admins').fetchone()[0]
|
|
conn.close()
|
|
|
|
if count == 0:
|
|
hash_pass = generate_password_hash(DEFAULT_PASS)
|
|
database.add_admin(DEFAULT_ADMIN, hash_pass)
|
|
logging.info(f"Default admin created: {DEFAULT_ADMIN}")
|
|
|
|
ensure_admin_exists()
|
|
|
|
# Настройка логирования
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format='%(asctime)s [%(levelname)s] %(message)s',
|
|
handlers=[logging.StreamHandler(sys.stdout)]
|
|
)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# --- Загрузка конфигурации ---
|
|
CONFIG_PATH = os.path.join(os.path.dirname(__file__), 'config.json')
|
|
|
|
def load_config():
|
|
"""Загрузка настроек из config.json и переменных окружения"""
|
|
config = {
|
|
"mikrotik": {
|
|
"host": "10.0.0.1",
|
|
"username": "admin",
|
|
"password": "password",
|
|
"port": 8728,
|
|
"port_ssl": 8729,
|
|
"use_ssl": False
|
|
},
|
|
"api": {
|
|
"port": 5000,
|
|
"debug": True
|
|
}
|
|
}
|
|
|
|
# 1. Загрузка из config.json
|
|
if os.path.exists(CONFIG_PATH):
|
|
try:
|
|
with open(CONFIG_PATH, 'r') as f:
|
|
user_config = json.load(f)
|
|
if 'mikrotik' in user_config:
|
|
config['mikrotik'].update(user_config['mikrotik'])
|
|
if 'api' in user_config:
|
|
config['api'].update(user_config['api'])
|
|
logger.info(f"Configuration loaded from {CONFIG_PATH}")
|
|
except Exception as e:
|
|
logger.error(f"Failed to load {CONFIG_PATH}: {e}")
|
|
|
|
# 2. Приоритет переменных окружения
|
|
config['mikrotik']['host'] = os.environ.get('MT_HOST', config['mikrotik']['host'])
|
|
config['mikrotik']['username'] = os.environ.get('MT_USER', config['mikrotik']['username'])
|
|
config['mikrotik']['password'] = os.environ.get('MT_PASS', config['mikrotik']['password'])
|
|
config['mikrotik']['use_ssl'] = os.environ.get('MT_SECURE', str(config['mikrotik']['use_ssl'])).lower() == 'true'
|
|
|
|
if config['mikrotik']['use_ssl']:
|
|
config['mikrotik']['port_active'] = int(os.environ.get('MT_PORT', config['mikrotik']['port_ssl']))
|
|
else:
|
|
config['mikrotik']['port_active'] = int(os.environ.get('MT_PORT', config['mikrotik']['port']))
|
|
|
|
return config
|
|
|
|
APP_CONFIG = load_config()
|
|
|
|
# --- Настройки безопасности ---
|
|
SECRET_KEY = os.environ.get('APP_SECRET_KEY', 'super-secret-key-change-me')
|
|
|
|
# --- Вспомогательные функции ---
|
|
def get_client_ip():
|
|
"""Определение IP адреса клиента с учетом прокси"""
|
|
if request.headers.get('X-Forwarded-For'):
|
|
return request.headers.get('X-Forwarded-For').split(',')[0].strip()
|
|
return request.remote_addr
|
|
|
|
class MikroTikAPIError(Exception):
|
|
"""Базовое исключение для MikroTik API"""
|
|
pass
|
|
|
|
class MikroTikConnectionError(MikroTikAPIError):
|
|
"""Ошибка при установке соединения"""
|
|
pass
|
|
|
|
class MikroTikAPI:
|
|
"""Класс для взаимодействия с MikroTik RouterOS API"""
|
|
|
|
def __init__(self, host, username, password, port=8728, secure=False, timeout=10):
|
|
self.host = host
|
|
self.username = username
|
|
self.password = password
|
|
self.port = port
|
|
self.secure = secure
|
|
self.timeout = timeout
|
|
self.connection = None
|
|
|
|
def connect(self):
|
|
try:
|
|
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
|
sock.settimeout(self.timeout)
|
|
if self.secure:
|
|
context = ssl.create_default_context()
|
|
context.check_hostname = False
|
|
context.verify_mode = ssl.CERT_NONE
|
|
self.connection = context.wrap_socket(sock, server_hostname=self.host)
|
|
else:
|
|
self.connection = sock
|
|
self.connection.connect((self.host, self.port))
|
|
logger.info(f"Connected to {self.host}:{self.port}")
|
|
result = self._talk(["/login", f"=name={self.username}", f"=password={self.password}"])
|
|
for reply, attrs in result:
|
|
if reply == '!trap':
|
|
error_msg = attrs.get('=message', 'Unknown login error')
|
|
raise MikroTikConnectionError(f"Login failed: {error_msg}")
|
|
return True
|
|
except Exception as e:
|
|
logger.error(f"Connection error: {e}")
|
|
raise MikroTikConnectionError(str(e))
|
|
|
|
def _write_len(self, length):
|
|
if length < 0x80: self.connection.sendall(length.to_bytes(1, 'big'))
|
|
elif length < 0x4000: self.connection.sendall((length | 0x8000).to_bytes(2, 'big'))
|
|
elif length < 0x200000: self.connection.sendall((length | 0xC00000).to_bytes(3, 'big'))
|
|
elif length < 0x10000000: self.connection.sendall((length | 0xE0000000).to_bytes(4, 'big'))
|
|
else:
|
|
self.connection.sendall((0xF0).to_bytes(1, 'big'))
|
|
self.connection.sendall(length.to_bytes(4, 'big'))
|
|
|
|
def _read_len(self):
|
|
byte = self.connection.recv(1)
|
|
if not byte: return 0
|
|
c = byte[0]
|
|
if (c & 0x80) == 0x00: return c
|
|
elif (c & 0xC0) == 0x80: return ((c & ~0xC0) << 8) + self.connection.recv(1)[0]
|
|
elif (c & 0xE0) == 0xC0: return ((c & ~0xE0) << 16) + int.from_bytes(self.connection.recv(2), 'big')
|
|
elif (c & 0xF0) == 0xE0: return ((c & ~0xF0) << 24) + int.from_bytes(self.connection.recv(3), 'big')
|
|
elif (c & 0xF8) == 0xF0: return int.from_bytes(self.connection.recv(4), 'big')
|
|
return 0
|
|
|
|
def _write_word(self, word):
|
|
encoded = word.encode('utf-8')
|
|
self._write_len(len(encoded))
|
|
self.connection.sendall(encoded)
|
|
|
|
def _read_word(self):
|
|
length = self._read_len()
|
|
if length == 0: return ''
|
|
data = b''
|
|
while len(data) < length:
|
|
chunk = self.connection.recv(length - len(data))
|
|
if not chunk: break
|
|
data += chunk
|
|
return data.decode('utf-8', errors='replace')
|
|
|
|
def _talk(self, words):
|
|
for word in words: self._write_word(word)
|
|
self._write_word('')
|
|
result = []
|
|
while True:
|
|
sentence = []
|
|
while True:
|
|
word = self._read_word()
|
|
if word == '': break
|
|
sentence.append(word)
|
|
if not sentence: continue
|
|
reply = sentence[0]
|
|
attrs = {}
|
|
for word in sentence[1:]:
|
|
eq_pos = word.find('=', 1)
|
|
if eq_pos == -1: attrs[word.lstrip('=.')] = ''
|
|
else: attrs[word[1:eq_pos]] = word[eq_pos+1:]
|
|
result.append((reply, attrs))
|
|
if reply == '!done': break
|
|
if reply == '!trap': break
|
|
return result
|
|
|
|
def get_address_list(self, list_name=None):
|
|
cmd = ["/ip/firewall/address-list/print"]
|
|
if list_name: cmd.append(f"?list={list_name}")
|
|
result = self._talk(cmd)
|
|
return [{ 'id': a.get('.id'), 'address': a.get('address'), 'list': a.get('list'), 'disabled': a.get('disabled') == 'true' } for r, a in result if r == '!re']
|
|
|
|
def add_address(self, address, list_name, comment=""):
|
|
cmd = ["/ip/firewall/address-list/add", f"=address={address}", f"=list={list_name}"]
|
|
if comment: cmd.append(f"=comment={comment}")
|
|
result = self._talk(cmd)
|
|
for r, a in result:
|
|
if r == '!trap': return False, a.get('message', 'Unknown error')
|
|
if r == '!done': return True, a.get('ret')
|
|
return False, "Failed"
|
|
|
|
def set_address_state(self, address_id, disabled=False):
|
|
cmd = ["/ip/firewall/address-list/set", f"=.id={address_id}", f"=disabled={'yes' if disabled else 'no'}"]
|
|
result = self._talk(cmd)
|
|
for r, a in result:
|
|
if r == '!trap': return False, a.get('message', 'Unknown error')
|
|
return True, None
|
|
|
|
def remove_address(self, address_id):
|
|
cmd = ["/ip/firewall/address-list/remove", f"=.id={address_id}"]
|
|
result = self._talk(cmd)
|
|
for r, a in result:
|
|
if r == '!trap': return False, a.get('message', 'Unknown error')
|
|
return True, None
|
|
|
|
def close(self):
|
|
if self.connection:
|
|
try: self.connection.close()
|
|
except: pass
|
|
self.connection = None
|
|
|
|
app = Flask(__name__)
|
|
|
|
@contextmanager
|
|
def get_mt_api():
|
|
mt = APP_CONFIG['mikrotik']
|
|
api = MikroTikAPI(mt['host'], mt['username'], mt['password'], port=mt['port_active'], secure=mt['use_ssl'])
|
|
try:
|
|
api.connect()
|
|
yield api
|
|
finally:
|
|
api.close()
|
|
|
|
def response_json(success, data=None, error=None, status=200):
|
|
resp = {"success": success}
|
|
if data is not None: resp["data"] = data
|
|
if error is not None: resp["error"] = error
|
|
return jsonify(resp), status
|
|
|
|
# --- Authentication Helpers ---
|
|
|
|
def token_required(f):
|
|
@wraps(f)
|
|
def decorated(*args, **kwargs):
|
|
token = None
|
|
if 'Authorization' in request.headers:
|
|
auth_header = request.headers['Authorization']
|
|
if auth_header.startswith('Bearer '):
|
|
token = auth_header.split(" ")[1]
|
|
|
|
if not token:
|
|
return response_json(False, error="Token is missing", status=401)
|
|
|
|
try:
|
|
data = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
|
|
g.admin_user = data['user']
|
|
except Exception:
|
|
return response_json(False, error="Token is invalid or expired", status=401)
|
|
|
|
return f(*args, **kwargs)
|
|
return decorated
|
|
|
|
@app.route('/api/v1/auth/login', methods=['POST'])
|
|
def login():
|
|
auth = request.json
|
|
username = auth.get('username')
|
|
password = auth.get('password')
|
|
|
|
if not username or not password:
|
|
return response_json(False, error="Username and Password required", status=400)
|
|
|
|
admin = database.get_admin(username)
|
|
if admin and check_password_hash(admin['password_hash'], password):
|
|
token = jwt.encode({
|
|
'user': username,
|
|
'exp': datetime.datetime.utcnow() + datetime.timedelta(hours=24)
|
|
}, SECRET_KEY)
|
|
return response_json(True, data={'token': token})
|
|
|
|
return response_json(False, error="Invalid credentials", status=401)
|
|
|
|
@app.route('/api/v1/admin/profile', methods=['POST'])
|
|
@token_required
|
|
def update_profile():
|
|
data = request.json
|
|
new_username = data.get('username')
|
|
new_password = data.get('password')
|
|
|
|
if not new_username:
|
|
return response_json(False, error="Username is required", status=400)
|
|
|
|
# Хешируем новый пароль, если он предоставлен
|
|
admin = database.get_admin(g.admin_user)
|
|
new_hash = admin['password_hash']
|
|
if new_password:
|
|
new_hash = generate_password_hash(new_password)
|
|
|
|
if database.update_admin(g.admin_user, new_username, new_hash):
|
|
# Если имя сменилось, нужно будет перелогиниться или выдать новый токен
|
|
return response_json(True, data={"message": "Profile updated successfully"})
|
|
return response_json(False, error="Failed to update profile", status=500)
|
|
|
|
@app.errorhandler(MikroTikAPIError)
|
|
def handle_api_error(e): return response_json(False, error=str(e), status=502)
|
|
|
|
@app.errorhandler(Exception)
|
|
def handle_generic_error(e):
|
|
if isinstance(e, HTTPException):
|
|
return response_json(False, error=e.description, status=e.code)
|
|
logger.exception("Unhandled server error")
|
|
return response_json(False, error="Internal Server Error", status=500)
|
|
|
|
@app.route('/api/v1/addresses', methods=['GET'])
|
|
@token_required
|
|
def list_addresses():
|
|
return response_json(True, data=database.get_all_items(request.args.get('list')))
|
|
|
|
@app.route('/api/v1/lists', methods=['GET'])
|
|
@token_required
|
|
def list_names():
|
|
# Возвращает простой список строк для выпадающих списков
|
|
return response_json(True, data=database.get_unique_lists())
|
|
|
|
@app.route('/api/v1/named-lists', methods=['GET'])
|
|
@token_required
|
|
def get_named_lists():
|
|
# Возвращает подробный список с ID для управления
|
|
return response_json(True, data=database.get_named_lists_detailed())
|
|
|
|
@app.route('/api/v1/named-lists', methods=['POST'])
|
|
@token_required
|
|
def create_named_list():
|
|
data = request.json
|
|
name = data.get('name')
|
|
if not name:
|
|
return response_json(False, error="Name is required", status=400)
|
|
|
|
success, error = database.add_named_list(name)
|
|
if success:
|
|
return response_json(True, status=201)
|
|
return response_json(False, error=error, status=400)
|
|
|
|
@app.route('/api/v1/named-lists/<int:list_id>', methods=['DELETE'])
|
|
@token_required
|
|
def delete_named_list(list_id):
|
|
success, error = database.delete_named_list(list_id)
|
|
if success:
|
|
return response_json(True)
|
|
return response_json(False, error=error, status=400)
|
|
|
|
@app.route('/api/v1/addresses', methods=['POST'])
|
|
@token_required
|
|
def create_address():
|
|
req = request.json or {}
|
|
address, list_name, comment = req.get('address'), req.get('list', 'default'), req.get('comment', '')
|
|
if not address: return response_json(False, error="Address required", status=400)
|
|
with get_mt_api() as api:
|
|
success, res = api.add_address(address, list_name, comment)
|
|
if success:
|
|
database.add_item(address, list_name, comment, res)
|
|
return response_json(True, data={"mikrotik_id": res}, status=201)
|
|
return response_json(False, error=res, status=400)
|
|
|
|
@app.route('/api/v1/addresses/<int:db_id>', methods=['DELETE'])
|
|
@token_required
|
|
def delete_address(db_id):
|
|
item = database.delete_item_by_id(db_id)
|
|
if not item: return response_json(False, error="Not found", status=404)
|
|
with get_mt_api() as api:
|
|
api.remove_address(item['mikrotik_id'])
|
|
return response_json(True)
|
|
|
|
@app.route('/api/v1/addresses/<int:db_id>/enable', methods=['PUT'])
|
|
@token_required
|
|
def enable_address(db_id):
|
|
item = database.get_item_by_id(db_id)
|
|
if not item: return response_json(False, error="Not found", status=404)
|
|
with get_mt_api() as api:
|
|
success, error = api.set_address_state(item['mikrotik_id'], disabled=False)
|
|
if success:
|
|
database.update_item_status(item['ip'], enabled=True)
|
|
return response_json(True)
|
|
return response_json(False, error=error, status=400)
|
|
|
|
@app.route('/api/v1/addresses/<int:db_id>/disable', methods=['PUT'])
|
|
@token_required
|
|
def disable_address(db_id):
|
|
item = database.get_item_by_id(db_id)
|
|
if not item: return response_json(False, error="Not found", status=404)
|
|
with get_mt_api() as api:
|
|
success, error = api.set_address_state(item['mikrotik_id'], disabled=True)
|
|
if success:
|
|
database.update_item_status(item['ip'], enabled=False)
|
|
return response_json(True)
|
|
return response_json(False, error=error, status=400)
|
|
|
|
@app.route('/')
|
|
def index():
|
|
"""Служит основной дашборд"""
|
|
return render_template('index.html')
|
|
|
|
@app.route('/admin')
|
|
def admin():
|
|
"""Служит админ-панель"""
|
|
return render_template('admin.html')
|
|
|
|
@app.route('/login')
|
|
def login_page():
|
|
"""Служит страницу входа"""
|
|
return render_template('login.html')
|
|
|
|
@app.route('/api/v1/client/status', methods=['GET'])
|
|
def client_status():
|
|
ip = get_client_ip()
|
|
item = database.get_item_by_ip(ip)
|
|
if not item: return response_json(False, error="Нет интеграции с сервисом", status=403)
|
|
return response_json(True, data={"ip": ip, "enabled": bool(item['enabled']), "list": item['list_name']})
|
|
|
|
@app.route('/api/v1/client/toggle', methods=['PUT'])
|
|
def client_toggle():
|
|
ip = get_client_ip()
|
|
item = database.get_item_by_ip(ip)
|
|
if not item: return response_json(False, error="Access denied", status=403)
|
|
new_state = not bool(item['enabled'])
|
|
with get_mt_api() as api:
|
|
success, error = api.set_address_state(item['mikrotik_id'], disabled=not new_state)
|
|
if success:
|
|
database.update_item_status(ip, enabled=new_state)
|
|
return response_json(True, data={"enabled": new_state})
|
|
return response_json(False, error=error, status=400)
|
|
|
|
@app.route('/health', methods=['GET'])
|
|
def health(): return response_json(True, data={"status": "ok"})
|
|
|
|
if __name__ == '__main__':
|
|
app.run(host='0.0.0.0', port=APP_CONFIG['api']['port'], debug=APP_CONFIG['api']['debug']) |