A minimal, easy-to-understand endpoint management demo:
- Single FastAPI server with SQLite (no Docker, no brokers).
- Simple Python agent that enrolls, heartbeats, pulls tasks, and uploads results.
- Shared-secret auth; optional HTTPS.
- Windows or Linux with Python 3.11+
- Recommended: create a virtual environment
python -m venv .venv
. .venv/Scripts/Activate.ps1 # Windows PowerShell
pip install -r server/requirements.txt
# Option A: use helper script
./run_server.ps1
# Option B: run as module
python -m server.main
Server listens on http://127.0.0.1:8000 by default. Open http://127.0.0.1:8000/docs for Swagger.
- Edit
agent/config.json(server URL and agent name).
pip install -r agent/requirements.txt
# Option A: use helper script
./run_agent.ps1
# Option B: run as module
python -m agent.agent
This prototype lets the agent run commands remotely via a new task type exec.
- Disabled controls: no allowlist yet (you will add it later). Use only in a trusted lab.
- Safety in place: no shell execution, default timeout 10s, output capped to 16KB.
Example task payloads (create via Swagger POST /tasks, header X-Admin-Token: changeme-admin):
- Windows PowerShell (use
cmd /c):
{
"target_agent_id": "agent-001",
"type": "exec",
"payload": { "cmd": "cmd", "args": ["/c", "echo", "hello"] }
}
- Linux/macOS:
{
"target_agent_id": "agent-001",
"type": "exec",
"payload": { "cmd": "echo", "args": ["hello"] }
}
Result will include: returncode, stdout, stderr, duration_ms, and truncation flags.
Current task type values supported by the agent (agent/agent.py dispatcher):
- inventory
- metrics
- logs
- exec
- processes_list
- network_info
- open_ports
- uptime
- user_sessions
- disk_usage_detail
Example payloads (admin → POST /tasks):
- processes_list
{
"target_agent_id": "agent-001",
"type": "processes_list",
"payload": { "name_filter": "python", "limit": 100 }
}
- open_ports
{
"target_agent_id": "agent-001",
"type": "open_ports",
"payload": { "protocols": ["tcp"], "listening_only": true, "limit": 100 }
}
- uptime
{
"target_agent_id": "agent-001",
"type": "uptime",
"payload": {}
}
- user_sessions
{
"target_agent_id": "agent-001",
"type": "user_sessions",
"payload": {}
}
- disk_usage_detail
{
"target_agent_id": "agent-001",
"type": "disk_usage_detail",
"payload": {}
}
You can run the server over HTTPS and have the agent verify a specific certificate (pinning).
- Generate a self-signed certificate (PowerShell example):
openssl req -x509 -newkey rsa:2048 -nodes -keyout server.key -out server.crt -days 365 -subj "/CN=127.0.0.1"
- Run the server with TLS:
uvicorn server.main:app --host 127.0.0.1 --port 8443 --ssl-keyfile server.key --ssl-certfile server.crt
- Compute the SHA256 pin of the certificate:
openssl x509 -in server.crt -noout -fingerprint -sha256
# Or in pure DER digest:
openssl x509 -in server.crt -outform der | openssl dgst -sha256
Use the hex digest (without spaces/colons) as pin_sha256.
- Configure the agent
agent/config.json:
{
"server_url": "https://127.0.0.1:8443",
"agent_id": "agent-001",
...,
"ca_cert_path": "C:/path/to/server.crt", // use full path
"pin_sha256": "<hex_sha256_of_cert_der>"
}
- Run the agent. It will verify the certificate using the CA file and enforce the SHA256 pin.
- Open http://127.0.0.1:8000/docs
- Use
POST /taskswith admin token (default:changeme-admininserver/config.py). - Example payload to ask an agent to send inventory:
{
"target_agent_id": "<AGENT_ID>",
"type": "inventory",
"payload": {}
}
The agent will poll tasks on heartbeat and post results.
server/FastAPI app and SQLite modelsagent/Python agent and simple modulesREADME.mdthis file.gitignore
server/config.pyholds admin token and DB path.- Each agent gets its own token on enrollment. Agents authenticate with header
X-Agent-Token.
- This is a minimal demo. For production:
- Use HTTPS (TLS certs) and rotate tokens.
- Add RBAC and better auditing.
MIT