#!/usr/bin/env python3
import os
import socket
import subprocess
import uuid
from flask import Flask, request, render_template_string, flash
app = Flask(__name__)
app.secret_key = "change-me-in-production"
# --- Config -----------------------------------------------------------------
DOCKER_IMAGE = "micro-debian-dev:latest" # built from the Dockerfile above
DATA_ROOT = "/srv/microcontainers" # where per-tenant dirs live
HOSTNAME = "arthur.lan" # or "mentalnet.xyz" if you prefer
PORT_RANGE_START = 20000
PORT_RANGE_END = 21000
os.makedirs(DATA_ROOT, exist_ok=True)
# --- Helpers ----------------------------------------------------------------
def find_free_port(start=PORT_RANGE_START, end=PORT_RANGE_END):
"""Find a free TCP port on the host in the given range."""
for port in range(start, end):
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.settimeout(0.2)
try:
s.bind(("", port))
return port
except OSError:
continue
raise RuntimeError("No free ports available in range")
def validate_ssh_key(key: str) -> bool:
"""Very basic SSH public key validation."""
key = key.strip()
if not key:
return False
allowed_prefixes = (
"ssh-ed25519",
"ssh-rsa",
"ecdsa-sha2-",
"sk-ssh-",
)
return any(key.startswith(p) for p in allowed_prefixes)
def run_docker_container(container_name: str, ssh_dir: str, host_port: int):
"""
Start the tenant container:
- Binds host_port -> container:22
- Mounts authorized_keys -> /home/micro/.ssh/authorized_keys (read-only)
"""
auth_keys_path = os.path.join(ssh_dir, "authorized_keys")
cmd = [
"docker", "run", "-d",
"--name", container_name,
"--memory=512m",
"--memory-swap=512m",
"-p", f"{host_port}:22",
"-v", f"{auth_keys_path}:/home/micro/.ssh/authorized_keys:ro",
DOCKER_IMAGE,
]
subprocess.run(cmd, check=True)
# --- Template ---------------------------------------------------------------
INDEX_TEMPLATE = """
MicroContainer Provisioning
MicroContainer Provisioning (Prototype)
This prototype spins up a small Docker-based "micro VM" and wires in your SSH public key.
- Base image:
{{ docker_image }}
- User:
micro
- SSH inside container:
port 22
- Host port: auto-assigned from
{{ port_start }}-{{ port_end }}
{% with messages = get_flashed_messages() %}
{% if messages %}
{% for msg in messages %}
{{ msg }}
{% endfor %}
{% endif %}
{% endwith %}
{% if result %}
Container created!
Container name: {{ result.name }}
Host port: {{ result.port }}
SSH command:
ssh micro@{{ hostname }} -p {{ result.port }}
You can change the password for micro after logging in with:
passwd
{% endif %}
"""
# --- Routes -----------------------------------------------------------------
@app.route("/", methods=["GET", "POST"])
def index():
if request.method == "GET":
return render_template_string(
INDEX_TEMPLATE,
docker_image=DOCKER_IMAGE,
port_start=PORT_RANGE_START,
port_end=PORT_RANGE_END,
result=None,
ssh_key="",
hostname=HOSTNAME,
)
# POST: handle provisioning
ssh_key = (request.form.get("ssh_key") or "").strip()
note = (request.form.get("note") or "").strip()
if not validate_ssh_key(ssh_key):
flash("Invalid SSH public key format. Please paste a standard OpenSSH public key line.")
return render_template_string(
INDEX_TEMPLATE,
docker_image=DOCKER_IMAGE,
port_start=PORT_RANGE_START,
port_end=PORT_RANGE_END,
result=None,
ssh_key=ssh_key,
hostname=HOSTNAME,
)
# Generate a container name
suffix = uuid.uuid4().hex[:6]
base_name = "mc"
if note:
safe_note = "".join(c for c in note.lower().replace(" ", "-") if c.isalnum() or c in "-_")
base_name = f"mc-{safe_note}"
container_name = f"{base_name}-{suffix}"
# Tenant SSH dir on host
tenant_root = os.path.join(DATA_ROOT, container_name)
tenant_ssh_dir = os.path.join(tenant_root, "ssh")
os.makedirs(tenant_ssh_dir, exist_ok=True)
# Write authorized_keys
auth_keys_path = os.path.join(tenant_ssh_dir, "authorized_keys")
with open(auth_keys_path, "w") as f:
f.write(ssh_key.strip() + "\n")
# Make it look like micro:micro (uid/gid 1000) to sshd in the container
os.chmod(auth_keys_path, 0o600)
try:
os.chown(auth_keys_path, 1000, 1000) # micro:micro inside container
except PermissionError:
# If Flask isn't running as root, this will fail; on your VM it should succeed.
pass
# Choose a free port and start container
host_port = find_free_port()
try:
run_docker_container(container_name, tenant_ssh_dir, host_port)
except subprocess.CalledProcessError as e:
flash(f"Error starting container: {e}")
return render_template_string(
INDEX_TEMPLATE,
docker_image=DOCKER_IMAGE,
port_start=PORT_RANGE_START,
port_end=PORT_RANGE_END,
result=None,
ssh_key=ssh_key,
hostname=HOSTNAME,
)
class Result:
def __init__(self, name, port):
self.name = name
self.port = port
result = Result(container_name, host_port)
return render_template_string(
INDEX_TEMPLATE,
docker_image=DOCKER_IMAGE,
port_start=PORT_RANGE_START,
port_end=PORT_RANGE_END,
result=result,
ssh_key="",
hostname=HOSTNAME,
)
if __name__ == "__main__":
# For dev/testing only
app.run(host="0.0.0.0", port=5000, debug=True)