CroneKorkN
730625e36c
every libs/*.py and hooks/*.py now starts with a one-line module docstring; every bin/* script starts with a `# purpose:` header. discovery-by-`ls`-and-read instead of by index. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
133 lines
No EOL
3.4 KiB
Python
Executable file
133 lines
No EOL
3.4 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
# purpose: upsert one 1Password login per `bundle:routeros` node, keyed on the bw node id.
|
|
|
|
from bundlewrap.repo import Repository
|
|
from os.path import realpath, dirname
|
|
|
|
import json
|
|
import os
|
|
import subprocess
|
|
from dataclasses import dataclass
|
|
from typing import Optional, List
|
|
|
|
bw = Repository(dirname(dirname(realpath(__file__))))
|
|
|
|
VAULT=bw.vault.decrypt('encrypt$gAAAAABpLgX_xxb5NmNCl3cgHM0JL65GT6PHVXO5gwly7IkmWoEgkCDSuAcSAkNFB8Tb4RdnTdpzVQEUL1XppTKVto_O7_b11GjATiyQYiSfiQ8KZkTKLvk=').value
|
|
BW_TAG = "bw"
|
|
BUNDLEWRAP_FIELD_LABEL = "bundlewrap node id"
|
|
|
|
|
|
@dataclass
|
|
class OpResult:
|
|
stdout: str
|
|
stderr: str
|
|
returncode: int
|
|
|
|
|
|
def main():
|
|
for node in bw.nodes_in_group('routeros'):
|
|
upsert_node_item(
|
|
node_name=node.name,
|
|
node_uuid=node.metadata.get('id'),
|
|
username=node.username,
|
|
password=node.password,
|
|
url=f'http://{node.hostname}',
|
|
)
|
|
|
|
|
|
def run_op(args):
|
|
proc = subprocess.run(
|
|
["op", "--vault", VAULT] + args,
|
|
env=os.environ.copy(),
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
|
|
if proc.returncode != 0:
|
|
raise RuntimeError(
|
|
f"op {' '.join(args)} failed with code {proc.returncode}:\n"
|
|
f"STDOUT:\n{proc.stdout}\n\nSTDERR:\n{proc.stderr}"
|
|
)
|
|
|
|
return OpResult(stdout=proc.stdout, stderr=proc.stderr, returncode=proc.returncode)
|
|
|
|
|
|
def op_item_list_bw():
|
|
out = run_op([
|
|
"item", "list",
|
|
"--tags", BW_TAG,
|
|
"--format", "json",
|
|
])
|
|
stdout = out.stdout.strip()
|
|
return json.loads(stdout) if stdout else []
|
|
|
|
|
|
def op_item_get(item_id):
|
|
args = ["item", "get", item_id, "--format", "json"]
|
|
return json.loads(run_op(args).stdout)
|
|
|
|
|
|
def op_item_create(title, node_uuid, username, password, url):
|
|
print(f"creating {title}")
|
|
return json.loads(run_op([
|
|
"item", "create",
|
|
"--category", "LOGIN",
|
|
"--title", title,
|
|
"--tags", BW_TAG,
|
|
"--url", url,
|
|
"--format", "json",
|
|
f"username={username}",
|
|
f"password={password}",
|
|
f"{BUNDLEWRAP_FIELD_LABEL}[text]={node_uuid}",
|
|
]).stdout)
|
|
|
|
|
|
def op_item_edit(item_id, title, username, password, url):
|
|
print(f"updating {title}")
|
|
return json.loads(run_op([
|
|
"item", "edit",
|
|
item_id,
|
|
"--title", title,
|
|
"--url", url,
|
|
"--format", "json",
|
|
f"username={username}",
|
|
f"password={password}",
|
|
]).stdout)
|
|
|
|
|
|
def find_node_item_id(node_uuid):
|
|
for summary in op_item_list_bw():
|
|
item_id = summary.get("id")
|
|
if not item_id:
|
|
continue
|
|
|
|
item = op_item_get(item_id)
|
|
for field in item.get("fields") or []:
|
|
label = field.get("label")
|
|
value = field.get("value")
|
|
if label == BUNDLEWRAP_FIELD_LABEL and value == node_uuid:
|
|
return item_id
|
|
return None
|
|
|
|
|
|
def upsert_node_item(node_name, node_uuid, username, password, url):
|
|
if item_id := find_node_item_id(node_uuid):
|
|
return op_item_edit(
|
|
item_id=item_id,
|
|
title=node_name,
|
|
username=username,
|
|
password=password,
|
|
url=url,
|
|
)
|
|
else:
|
|
return op_item_create(
|
|
title=node_name,
|
|
node_uuid=node_uuid,
|
|
username=username,
|
|
password=password,
|
|
url=url,
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main() |