dinlo
f624fee5e0
- SSH/SFTP управление без агентов - Пакеты (apt), Docker, файлы, systemd, ufw, cron - Плейбуки, групповые операции, планировщик - Бэкапы, отчёты, Portainer, Proxmox VE, Gitea Deploy - Терминал, Telegram-уведомления, тёмная/светлая тема
99 lines
3.2 KiB
Python
99 lines
3.2 KiB
Python
"""UFW файрвол."""
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from typing import List, Optional
|
|
|
|
from app.core.ssh import SSHClient
|
|
|
|
|
|
@dataclass
|
|
class Rule:
|
|
number: int
|
|
to: str
|
|
action: str
|
|
from_: str
|
|
raw: str
|
|
|
|
|
|
async def is_installed(c: SSHClient) -> bool:
|
|
r = await c.run("which ufw")
|
|
return r.ok and bool(r.stdout.strip())
|
|
|
|
|
|
async def status(c: SSHClient) -> tuple[bool, str]:
|
|
"""(active, raw_status)"""
|
|
r = await c.sudo("ufw status verbose", timeout=30)
|
|
active = "Status: active" in r.stdout
|
|
return active, r.stdout
|
|
|
|
|
|
async def list_rules(c: SSHClient) -> List[Rule]:
|
|
r = await c.sudo("ufw status numbered", timeout=30)
|
|
rules: List[Rule] = []
|
|
for line in r.stdout.splitlines():
|
|
line = line.strip()
|
|
if line.startswith("[") and "]" in line:
|
|
num = int(line[1:line.index("]")].strip())
|
|
rest = line[line.index("]") + 1:].strip()
|
|
parts = rest.split()
|
|
# "22/tcp ALLOW IN Anywhere"
|
|
to = parts[0] if parts else ""
|
|
action = " ".join(parts[1:3]) if len(parts) > 2 else (parts[1] if len(parts) > 1 else "")
|
|
from_ = " ".join(parts[3:]) if len(parts) > 3 else ""
|
|
rules.append(Rule(num, to, action, from_, line))
|
|
return rules
|
|
|
|
|
|
async def ensure_ssh_allowed(c: SSHClient, ssh_port: int) -> None:
|
|
"""Защита от отстрела себя: разрешить текущий SSH-порт."""
|
|
rules = await list_rules(c)
|
|
for rule in rules:
|
|
if str(ssh_port) in rule.to and "ALLOW" in rule.action.upper():
|
|
return
|
|
await c.sudo(f"ufw allow {ssh_port}/tcp", timeout=30)
|
|
|
|
|
|
async def enable(c: SSHClient, ssh_port: int) -> str:
|
|
await ensure_ssh_allowed(c, ssh_port)
|
|
r = await c.sudo("yes | ufw enable", timeout=30)
|
|
return r.stdout + r.stderr
|
|
|
|
|
|
async def disable(c: SSHClient) -> str:
|
|
r = await c.sudo("ufw disable", timeout=30)
|
|
return r.stdout + r.stderr
|
|
|
|
|
|
async def add_rule(c: SSHClient, port: str, action: str = "allow",
|
|
proto: str = "tcp", from_addr: str = "") -> str:
|
|
assert action in ("allow", "deny", "limit", "reject")
|
|
src = f"from {from_addr} " if from_addr else ""
|
|
r = await c.sudo(f"ufw {action} {src}to any port {port} proto {proto}", timeout=30)
|
|
return r.stdout + r.stderr
|
|
|
|
|
|
async def delete_rule(c: SSHClient, number: int) -> str:
|
|
r = await c.sudo(f"yes | ufw delete {number}", timeout=30)
|
|
return r.stdout + r.stderr
|
|
|
|
|
|
async def listening_ports(c: SSHClient) -> List[str]:
|
|
r = await c.sudo("ss -tlnp | awk 'NR>1{print $4, $6}'", timeout=30)
|
|
return [l for l in r.stdout.splitlines() if l.strip()]
|
|
|
|
|
|
PRESETS = {
|
|
"Только SSH": [("{ssh_port}", "allow", "tcp")],
|
|
"Веб-сервер": [("{ssh_port}", "allow", "tcp"), ("80", "allow", "tcp"), ("443", "allow", "tcp")],
|
|
"Docker-хост": [("{ssh_port}", "allow", "tcp"), ("2375", "deny", "tcp"), ("2376", "deny", "tcp")],
|
|
}
|
|
|
|
|
|
async def apply_preset(c: SSHClient, preset_name: str, ssh_port: int) -> str:
|
|
log = []
|
|
for port, action, proto in PRESETS[preset_name]:
|
|
p = port.format(ssh_port=ssh_port)
|
|
log.append(await add_rule(c, p, action, proto))
|
|
return "\n".join(log)
|