#!/usr/bin/env python3
import asyncio
import shutil
import uuid
import os
import pty
import struct
import fcntl
import termios
import signal
import json
from pathlib import Path
from fastapi import FastAPI, WebSocket, HTTPException
from fastapi.responses import HTMLResponse, FileResponse, RedirectResponse
import uvicorn
app = FastAPI()
active_sessions = {}
@app.get("/")
async def root():
return HTMLResponse("""
<!DOCTYPE html>
<html>
<head>
<meta charset="UTF-8">
<title>Preparing Environment</title>
<style>
* { margin: 0; padding: 0; box-sizing: border-box;-ms-overflow-style: none;
scrollbar-width: none;
}
*::-webkit-scrollbar {
display: none;
}
body {
display: flex;
overflow: hidden;
align-items: center;
justify-content: center;
height: 100vh;
font-family: Arial, sans-serif;
background: #1e1e1e;
color: #fff;
}
.container { text-align: center; }
.spinner {
border: 4px solid rgba(255,255,255,0.1);
border-top: 4px solid #fff;
border-radius: 50%;
width: 50px;
height: 50px;
animation: spin 1s linear infinite;
margin: 0 auto 20px;
}
@keyframes spin { 0% { transform: rotate(0deg); } 100% { transform: rotate(360deg); } }
</style>
</head>
<body>
<div class="container">
<div class="spinner"></div>
<h1>Preparing Environment</h1>
<p>Please wait while we set up your workspace...</p>
</div>
<script>
fetch('/prepare').then(r => r.json()).then(data => {
window.location.href = '/terminal/' + data.session_id;
});
</script>
</body>
</html>
""")
@app.get("/prepare")
async def prepare():
session_id = str(uuid.uuid4())
system_directory = Path("system")
workspace_directory = Path(f"workspace/{session_id}")
if not system_directory.exists():
raise HTTPException(status_code=500, detail="System directory not found")
shutil.copytree(system_directory, workspace_directory)
active_sessions[session_id] = {"workspace": workspace_directory}
return {"session_id": session_id}
@app.get("/terminal/{session_id}")
async def terminal(session_id: str):
if session_id not in active_sessions:
return RedirectResponse(url="/")
return HTMLResponse(f"""
<!DOCTYPE html>
<html>
<head>
<meta charset="UTF-8">
<title>Terminal</title>
<link rel="stylesheet" href="https://cdn.jsdelivr.net/npm/xterm@5.3.0/css/xterm.css" />
<style>
* {{ margin: 0; padding: 0; box-sizing: border-box; }}
body {{ overflow: hidden; }}
#terminal {{ width: 100vw; height: 100vh; }}
</style>
</head>
<body>
<div id="terminal"></div>
<script src="https://cdn.jsdelivr.net/npm/xterm@5.3.0/lib/xterm.js"></script>
<script src="https://cdn.jsdelivr.net/npm/xterm-addon-fit@0.8.0/lib/xterm-addon-fit.js"></script>
<script>
const term = new Terminal({{
cursorBlink: true,
fontSize: 14,
theme: {{
background: '#1e1e1e',
foreground: '#f0f0f0'
}}
}});
const fitAddon = new FitAddon.FitAddon();
term.loadAddon(fitAddon);
term.open(document.getElementById('terminal'));
fitAddon.fit();
const ws = new WebSocket('ws://' + window.location.host + '/ws/{session_id}');
ws.onopen = () => {{
const dims = {{ cols: term.cols, rows: term.rows }};
ws.send(JSON.stringify({{ type: 'resize', data: dims }}));
}};
ws.onmessage = (event) => {{
term.write(event.data);
}};
ws.onclose = () => {{
window.location.href = window.location.href.replace('terminal', 'files');
term.writeln('\\r\\nThat went smooth huh, that\s sum Molodetz for ya!');
}};
term.onData((data) => {{
ws.send(JSON.stringify({{ type: 'input', data: data }}));
}});
term.onResize(({{ cols, rows }}) => {{
ws.send(JSON.stringify({{ type: 'resize', data: {{ cols, rows }} }}));
}});
window.addEventListener('resize', () => {{
fitAddon.fit();
}});
document.addEventListener('DOMContentLoaded', function() {{
const elements = document.querySelectorAll('.xterm .xterm-viewport');
elements.forEach(element => {{
element.style.setProperty('overflow-y', 'hidden', 'important');
}});
}});
</script>
</body>
</html>
""")
def set_winsize(fd, rows, cols):
winsize = struct.pack("HHHH", rows, cols, 0, 0)
fcntl.ioctl(fd, termios.TIOCSWINSZ, winsize)
@app.websocket("/ws/{session_id}")
async def websocket_endpoint(websocket: WebSocket, session_id: str):
if session_id not in active_sessions:
await websocket.close(code=1008)
return
await websocket.accept()
workspace_directory = active_sessions[session_id]["workspace"]
master_fd, slave_fd = pty.openpty()
event_loop = asyncio.get_event_loop()
process = await asyncio.create_subprocess_exec(
"./manage.py", "start",
stdin=slave_fd,
stdout=slave_fd,
stderr=slave_fd,
cwd=str(workspace_directory),
start_new_session=True
)
os.close(slave_fd)
async def read_output():
try:
while True:
data = await event_loop.run_in_executor(None, os.read, master_fd, 1024)
if not data:
break
await websocket.send_text(data.decode('utf-8', errors='replace'))
except Exception:
await websocket.close()
async def handle_input():
try:
while True:
message = await websocket.receive_text()
parsed_message = json.loads(message)
if parsed_message['type'] == 'input':
await event_loop.run_in_executor(None, os.write, master_fd, parsed_message['data'].encode('utf-8'))
elif parsed_message['type'] == 'resize':
set_winsize(master_fd, parsed_message['data']['rows'], parsed_message['data']['cols'])
except Exception:
await websocket.close()
try:
await asyncio.gather(read_output(), handle_input(), return_exceptions=True)
except Exception as e:
print(e)
await websocket.close()
try:
os.close(master_fd)
except Exception:
pass
try:
process.kill()
except Exception:
pass
await websocket.close()
@app.get("/files/{session_id}/{path:path}")
async def browse_files(session_id: str, path: str = ""):
if session_id not in active_sessions:
raise HTTPException(status_code=404, detail="Session not found")
workspace_directory = active_sessions[session_id]["workspace"]
target_path = (workspace_directory / path).resolve()
if not str(target_path).startswith(str(workspace_directory.resolve())):
raise HTTPException(status_code=403, detail="Access denied")
if not target_path.exists():
raise HTTPException(status_code=404, detail="Path not found")
if target_path.is_file():
return FileResponse(target_path)
items = []
for item in sorted(target_path.iterdir()):
rel_path = item.absolute().relative_to(workspace_directory.absolute())
items.append({
"name": item.name,
"is_dir": item.is_dir(),
"path": str(rel_path)
})
html = f"""
<!DOCTYPE html>
<html>
<head>
<meta charset="UTF-8">
<title>File Browser</title>
<style>
* {{ margin: 0; padding: 0; box-sizing: border-box; }}
body {{ font-family: Arial, sans-serif; padding: 20px; background: #f5f5f5; }}
h1 {{ margin-bottom: 20px; }}
.path {{ background: #fff; padding: 10px; margin-bottom: 20px; border-radius: 4px; border: 1px solid #ddd; }}
table {{ width: 100%; background: #fff; border-collapse: collapse; box-shadow: 0 2px 4px rgba(0,0,0,0.1); }}
th, td {{ padding: 12px; text-align: left; border-bottom: 1px solid #ddd; }}
th {{ background: #333; color: #fff; }}
a {{ color: #0066cc; text-decoration: none; }}
a:hover {{ text-decoration: underline; }}
.dir {{ font-weight: bold; }}
.file {{ color: #333; }}
</style>
</head>
<body>
<h1>File Browser</h1>
<div class="path">Path: /{path or ''}</div>
<table>
<thead>
<tr>
<th>Name</th>
<th>Type</th>
</tr>
</thead>
<tbody>
"""
if path:
parent = str(Path(path).parent) if Path(path).parent != Path('.') else ''
html += f'<tr><td><a href="/files/{session_id}/{parent}">..</a></td><td>Directory</td></tr>'
for item in items:
name_display = item['name'] + ('/' if item['is_dir'] else '')
item_type = 'Directory' if item['is_dir'] else 'File'
css_class = 'dir' if item['is_dir'] else 'file'
html += f'<tr><td><a class="{css_class}" href="/files/{session_id}/{item["path"]}">{name_display}</a></td><td>{item_type}</td></tr>'
html += """
</tbody>
</table>
</body>
</html>
"""
return HTMLResponse(html)
if __name__ == "__main__":
signal.signal(signal.SIGINT, lambda s, f: os._exit(0))
import argparse
parser = argparse.ArgumentParser(description="Run the application server.")
parser.add_argument("--port", type=int, default=8240, help="Port number to run the server on.")
arguments = parser.parse_args()
uvicorn.run(app, host="0.0.0.0", port=arguments.port)
Reference in New Issue View Git Blame Copy Permalink