Skip to content

Conversation

@captainpacket
Copy link

Adds a minimal netlab API server (stdlib HTTP) to expose CLI actions for automation. Includes new CLI command, help listing, and docs page.

@captainpacket
Copy link
Author

This is primarily an automation I built to hook netlab into Semaphore UI

@DanPartelly
Copy link
Collaborator

Thank you ! I certainly see some future in this.

Could you please write some words describing the problem you are trying to solve with this commit? So we can start from a common understanding?

@captainpacket
Copy link
Author

captainpacket commented Dec 24, 2025

Engineers here need to model and test lots of different vendors and networks. We've had some internal tooling in the past to do this manually (some hack expect style scripts), but have switched to netlab to make the environment a bit more composable.

We started using Semaphore UI (https://semaphoreui.com/) as a job execution engine, which works well with the built in terraform/ansible worklows in a multi-user environment. It can also do manual bash/python scripts. I'd started with bash scripts with sshpass, but found them kinda brittle. I wanted to switch to an API based workflow to make it a bit more robust.

I'd initially wrote a FastAPI based wrapper that imports these functions (which I'm still using), but I had the thought that this could be easily extended in the native CLI workflow. A typed version I think still has use and this could be extended for that.

@ipspace
Copy link
Owner

ipspace commented Dec 26, 2025

@captainpacket -- Thanks a million for the PR. Definitely an interesting idea ;) and I truly appreciate you crossing all the Ts and dotting all the Is.

Give me a few days -- I'm stuck in the "we have to deal with another Ansible brokenness" bog, and have to get out of that first.

@captainpacket captainpacket force-pushed the skyforge-api-server branch 5 times, most recently from 61a1f3d to 76d3284 Compare December 26, 2025 13:25
@captainpacket
Copy link
Author

Cleaned up a few semaphore specific things (project/task) that I'd left behind; added minimal authentication and TLS option

Copy link
Owner

@ipspace ipspace left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Mostly minor details and improvement suggestions, the only showstopper is whether jobs can be run in parallel and how to handle that.

netlab config <config.md>
netlab connect <connect.md>
netlab create <create.md>
netlab api <api.md>
Copy link
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think this list is sorted, so "netlab api" should be at the very top

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good catch — moved netlab api to the top so the list stays sorted.

from . import status as netlab_status
from . import up as netlab_up

DATA_DIR = os.getenv("NETLAB_API_DATA_DIR", "/var/lib/netlab/api")
Copy link
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't think this is user-writeable on most systems, so the HTTP server would have to be started as root. I usually use "/tmp/something" in these cases (or you could use "~/...")

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Agreed — changed the default data/log dir to a user-writable temp location; still overrideable with NETLAB_API_DATA_DIR.

handler.wfile.write(data)


def require_auth(handler: BaseHTTPRequestHandler) -> bool:
Copy link
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Just wondering: isn't there a Python library to handle this?

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I couldn’t find a clean stdlib helper for server-side Basic Auth parsing; kept stdlib-only to avoid extra deps and refactored it into a small helper to keep the handler clean.

def workspace_dir(payload: Dict[str, Any]) -> str:
workdir = (payload.get("workdir") or "").strip()
if workdir:
if os.path.isabs(workdir):
Copy link
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It looks to me like you're trying to get an absolute path to the working directory. If that's the case you could probably use payload.get("workdir",payload.get("workspaceRoot",os.getcwd())) to get the path you want to get or loop through the keywords instead of effectively having the same code twice.

Also, I'd usually use pathlib.Path(path).resolve() to get an absolute path.

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Refactored workdir/workspaceRoot selection and switched to pathlib.Path(...).resolve() for absolute paths; removed duplicated branches.



def resolve_topology(payload: Dict[str, Any], workdir: str) -> str:
topology_url = (payload.get("topologyUrl") or "").strip()
Copy link
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

As you're using the same expression so many times, it might make sense to turn it into a function

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done — factored the repeated path/payload handling into helper functions.

results: List[Dict[str, str]] = []
if not template_dir:
return results
for root, dirs, files in os.walk(template_dir):
Copy link
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It looks like you're doing list(glob('**/*.yml')) + list(glob('**/*yaml')). You could use glob from os.path or from pathlib

Copy link
Author

@captainpacket captainpacket Jan 2, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Switched template discovery to pathlib globbing for .yml/.yaml

_run_with_output(netlab_down.run, args)
elif action == "collect":
args = []
if payload.get("instance"):
Copy link
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I would probably do this with a dictionary mapping payload variable names into netlab args, and use the same approach for all commands.

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Agreed — refactored payload→argv construction to a consistent helper-based approach across actions.

os.chdir(prev_cwd)


def start_job(payload: Dict[str, Any]) -> Dict[str, Any]:
Copy link
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can you start jobs in parallel? If that's the case, you need multilab plugin, or you have to serialize the jobs.

Copy link
Author

@captainpacket captainpacket Jan 2, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good point; I’m serializing job execution (queueing requests but only running one netlab action at a time). If we later want true parallel jobs, we should detect/require multilab and unique ID per job rather than reinventing isolation.

send_json(self, HTTPStatus.NOT_FOUND, {"error": "not found"})
return

if parts == ["healthz"]:
Copy link
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I would usually implement something like this with a dictionary mapping keywords (parts in you case) into functions that do the actual work.

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Refactored routing to a dict mapping endpoints/keywords to handler functions (kept the /jobs//... special-case)



def run(cli_args: List[str]) -> None:
parser = argparse.ArgumentParser(description="netlab API server")
Copy link
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

All other CLI commands have parser in a separate function to keep the "run" function shorter (and cleaner)

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Aligned with other CLI modules — moved parser construction into a separate api_parse_args() function so run() stays short.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

3 participants