feat: OCI CLI Terminal, Spanish i18n, language in My Settings
Terminal: - New /terminal page with Linux-style terminal UI (dark theme, title bar, prompt) - OCI Config selector — isolated per user (ownership check) - Tab autocomplete: parses oci --help dynamically, caches results, commands first - Command history (↑/↓) persisted in DB per user - Security: only 'oci' commands, blocks pipes/redirects/shell injection - Built-in commands: clear, history, help - Timeout 60s per command i18n: - Spanish (es.ts): 748+ keys, Latin American formal - Language selector moved from sidebar to My Settings (🇧🇷/🇺🇸/🇪🇸) - Auto-detect browser language (pt/es/en) Backend: - POST /api/terminal/execute — run OCI CLI with selected config - GET /api/terminal/completions — dynamic autocomplete from oci --help - GET /api/terminal/history — per-user command history - terminal_history table in SQLite
This commit is contained in:
nogueiraguh
145
backend/app.py
145
backend/app.py
@@ -323,6 +323,12 @@ def init_db():
|
||||
action TEXT NOT NULL, resource TEXT, details TEXT,
|
||||
ip TEXT, created_at TEXT DEFAULT (datetime('now'))
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS terminal_history (
|
||||
id TEXT PRIMARY KEY, user_id TEXT NOT NULL,
|
||||
oci_config_id TEXT NOT NULL, command TEXT NOT NULL,
|
||||
exit_code INTEGER, output TEXT, error TEXT,
|
||||
created_at TEXT DEFAULT (datetime('now'))
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS config_logs (
|
||||
id TEXT PRIMARY KEY,
|
||||
config_type TEXT NOT NULL,
|
||||
@@ -1089,6 +1095,145 @@ async def test_oci(cid: str, u=Depends(require("admin","user"))):
|
||||
_config_log("oci", cid, cname, "error", "test", str(e)[:500], u["id"], u["username"])
|
||||
return {"status":"error","message":str(e)[:500]}
|
||||
|
||||
# ── OCI CLI Terminal ─────────────────────────────────────────────────────────
|
||||
_TERMINAL_BLOCKED_PATTERNS = re.compile(
|
||||
r'(rm\s|mv\s|cp\s|chmod\s|chown\s|sudo\s|bash\s|sh\s|curl\s|wget\s|python|pip\s|apt\s|yum\s|>\s|>>\s|\|)',
|
||||
re.IGNORECASE
|
||||
)
|
||||
|
||||
@app.post("/api/terminal/execute")
|
||||
async def terminal_execute(req: dict, u=Depends(current_user)):
|
||||
"""Execute an OCI CLI command using the selected OCI config. Only 'oci' commands allowed."""
|
||||
oci_config_id = req.get("oci_config_id", "")
|
||||
command = (req.get("command", "") or "").strip()
|
||||
if not oci_config_id:
|
||||
raise HTTPException(400, "oci_config_id obrigatório")
|
||||
if not command:
|
||||
raise HTTPException(400, "Comando vazio")
|
||||
_verify_config_access("oci", oci_config_id, u)
|
||||
# Security: only allow 'oci' commands
|
||||
if not command.startswith("oci "):
|
||||
raise HTTPException(400, "Apenas comandos 'oci' são permitidos. Ex: oci iam user list")
|
||||
if _TERMINAL_BLOCKED_PATTERNS.search(command):
|
||||
raise HTTPException(400, "Comando contém operações não permitidas")
|
||||
# Build config path
|
||||
cfg_path = OCI_DIR / oci_config_id / "config"
|
||||
if not cfg_path.exists():
|
||||
raise HTTPException(400, "OCI config file não encontrado. Reconfigure as credenciais.")
|
||||
# Execute with timeout
|
||||
import shlex, subprocess
|
||||
cmd_args = shlex.split(command) + ["--config-file", str(cfg_path)]
|
||||
try:
|
||||
loop = asyncio.get_event_loop()
|
||||
result = await asyncio.wait_for(
|
||||
loop.run_in_executor(_chat_executor, lambda: subprocess.run(
|
||||
cmd_args, capture_output=True, text=True, timeout=60, cwd="/tmp",
|
||||
env={**os.environ, "OCI_CLI_SUPPRESS_FILE_PERMISSIONS_WARNING": "True", "HOME": "/tmp"}
|
||||
)), timeout=65)
|
||||
output = result.stdout or ""
|
||||
error = result.stderr or ""
|
||||
exit_code = result.returncode
|
||||
# Save to history
|
||||
with db() as c:
|
||||
c.execute("INSERT INTO terminal_history (id,user_id,oci_config_id,command,exit_code,output,error) VALUES (?,?,?,?,?,?,?)",
|
||||
(str(uuid.uuid4()), u["id"], oci_config_id, command, exit_code, output[:10000], error[:2000]))
|
||||
return {
|
||||
"exit_code": exit_code,
|
||||
"output": output[:50000],
|
||||
"error": error[:5000] if exit_code != 0 else "",
|
||||
}
|
||||
except asyncio.TimeoutError:
|
||||
with db() as c:
|
||||
c.execute("INSERT INTO terminal_history (id,user_id,oci_config_id,command,exit_code,error) VALUES (?,?,?,?,?,?)",
|
||||
(str(uuid.uuid4()), u["id"], oci_config_id, command, -1, "Timeout: 60s"))
|
||||
return {"exit_code": -1, "output": "", "error": "Timeout: comando excedeu 60 segundos"}
|
||||
except Exception as e:
|
||||
return {"exit_code": -1, "output": "", "error": str(e)[:2000]}
|
||||
|
||||
@app.get("/api/terminal/history")
|
||||
async def terminal_history(oci_config_id: str = Query(""), limit: int = Query(50), u=Depends(current_user)):
|
||||
"""Get recent terminal commands for the user."""
|
||||
with db() as c:
|
||||
if oci_config_id:
|
||||
rows = c.execute(
|
||||
"SELECT * FROM terminal_history WHERE user_id=? AND oci_config_id=? ORDER BY created_at DESC LIMIT ?",
|
||||
(u["id"], oci_config_id, limit)).fetchall()
|
||||
else:
|
||||
rows = c.execute(
|
||||
"SELECT * FROM terminal_history WHERE user_id=? ORDER BY created_at DESC LIMIT ?",
|
||||
(u["id"], limit)).fetchall()
|
||||
return [dict(r) for r in rows]
|
||||
|
||||
# OCI CLI autocomplete cache
|
||||
_oci_completions_cache: dict = {}
|
||||
|
||||
def _parse_oci_help(args: list[str]) -> list[str]:
|
||||
"""Run oci <args> --help and extract available commands/options."""
|
||||
import subprocess
|
||||
try:
|
||||
result = subprocess.run(
|
||||
["oci"] + args + ["--help"], capture_output=True, text=True, timeout=10,
|
||||
env={**os.environ, "OCI_CLI_SUPPRESS_FILE_PERMISSIONS_WARNING": "True", "HOME": "/tmp"})
|
||||
text = result.stdout + result.stderr
|
||||
commands = []
|
||||
in_commands = False
|
||||
for line in text.splitlines():
|
||||
stripped = line.strip()
|
||||
# Detect transition into command/service listing
|
||||
if not in_commands:
|
||||
if re.match(r'^(Commands|Options|Core|Others|Networking|Database|Identity|Storage|Compute|Security)', stripped):
|
||||
in_commands = True
|
||||
continue
|
||||
if not stripped:
|
||||
continue
|
||||
# Section headers (single word or capitalized phrase not indented) — skip
|
||||
if not line.startswith(" "):
|
||||
continue
|
||||
# Capture " command-name Description text" pattern
|
||||
m = re.match(r'^ ([a-z][a-z0-9\-_]+)\s', line)
|
||||
if m and len(m.group(1)) > 1:
|
||||
commands.append(m.group(1))
|
||||
continue
|
||||
# Capture 2-space indent commands (sub-command help)
|
||||
m = re.match(r'^ ([a-z][a-z0-9\-_]+)\s', line)
|
||||
if m and len(m.group(1)) > 2:
|
||||
commands.append(m.group(1))
|
||||
continue
|
||||
# Capture --options
|
||||
if stripped.startswith("--"):
|
||||
opt = stripped.split()[0].rstrip(",")
|
||||
if opt.startswith("--"): commands.append(opt)
|
||||
return sorted(set(commands))
|
||||
except Exception:
|
||||
return []
|
||||
|
||||
@app.get("/api/terminal/completions")
|
||||
async def terminal_completions(prefix: str = Query(""), u=Depends(current_user)):
|
||||
"""Get OCI CLI autocomplete suggestions for a partial command."""
|
||||
parts = prefix.strip().split()
|
||||
if not parts or parts[0] != "oci":
|
||||
return {"suggestions": []}
|
||||
# Build cache key from command path (without final partial word)
|
||||
# e.g. "oci iam us" → lookup completions for ["oci", "iam"], filter by "us"
|
||||
if prefix.endswith(" "):
|
||||
cmd_parts = parts[1:]
|
||||
filter_prefix = ""
|
||||
else:
|
||||
cmd_parts = parts[1:-1]
|
||||
filter_prefix = parts[-1] if len(parts) > 1 else ""
|
||||
cache_key = " ".join(cmd_parts)
|
||||
if cache_key not in _oci_completions_cache:
|
||||
loop = asyncio.get_event_loop()
|
||||
completions = await loop.run_in_executor(_chat_executor, lambda: _parse_oci_help(cmd_parts))
|
||||
_oci_completions_cache[cache_key] = completions
|
||||
suggestions = _oci_completions_cache[cache_key]
|
||||
if filter_prefix:
|
||||
suggestions = [s for s in suggestions if s.startswith(filter_prefix)]
|
||||
# Commands first, then options
|
||||
cmds = [s for s in suggestions if not s.startswith("--")]
|
||||
opts = [s for s in suggestions if s.startswith("--")]
|
||||
return {"suggestions": (cmds + opts)[:50]}
|
||||
|
||||
# ── OCI Account Explorer ──────────────────────────────────────────────────────
|
||||
@app.get("/api/oci/explore/{cid}/compartments")
|
||||
async def explore_compartments(cid: str, u=Depends(current_user)):
|
||||
|
||||
Reference in New Issue
Block a user