Skip to content

Fix catalog ownership authorization #2

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
flemming-n-larsen merged 1 commit into from
Aug 3, 2026
Merged

Fix catalog ownership authorization #2

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
46 changes: 36 additions & 10 deletions scripts/validate_bot.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -139,27 +139,35 @@ def check_governance(bots: list[Bot], root: Path, owner: str) -> None:
disqualified_names = {entry["bot"] for entry in banned.get("disqualifiedBots", [])}
if owner in banned_accounts:
raise ValidationError(f"owner `{owner}` is banned from submissions")
known_names: dict[str, str] = {name: record["ownerId"] for record in owners.get("owners", []) for name in record.get("bots", [])}
catalog_entries = read_json(root / "bots" / "index.json").get("bots", []) if (root / "bots" / "index.json").exists() else []
catalog_by_name = {entry["name"]: entry for entry in catalog_entries if entry.get("status") == "active"}
submitted_bots = [
bot
for bot in bots
if (previous := catalog_by_name.get(bot.name)) is None
or previous.get("version") != bot.config["version"]
or previous.get("sourceHash") != bot.source_hash
]
owner_by_bot = {name: record for record in owners.get("owners", []) for name in record.get("bots", [])}
seen_skeletons: dict[str, str] = {}
for bot in bots:
for bot in submitted_bots:
if bot.name in disqualified_names:
raise ValidationError(f"bot `{bot.name}` is disqualified")
bot_skeleton = skeleton(bot.name)
previous = seen_skeletons.get(bot_skeleton)
if previous is not None and previous != bot.name:
raise ValidationError(f"bot `{bot.name}` is confusable with `{previous}`")
seen_skeletons[bot_skeleton] = bot.name
existing_owner = known_names.get(bot.name)
if existing_owner is not None and existing_owner != owner:
raise ValidationError(f"bot `{bot.name}` belongs to owner `{existing_owner}`")
existing_owner = owner_by_bot.get(bot.name)
if existing_owner is not None and owner not in existing_owner.get("accounts", []):
raise ValidationError(f"bot `{bot.name}` belongs to owner `{existing_owner['ownerId']}`")
previous = catalog_by_name.get(bot.name)
if previous is not None and previous.get("version") == bot.config["version"] and previous.get("sourceHash") != bot.source_hash:
raise ValidationError(f"bot `{bot.name}` changed source without increasing its version")
active_by_owner: dict[str, int] = {}
for bot in bots:
active_by_owner[known_names.get(bot.name, owner)] = active_by_owner.get(known_names.get(bot.name, owner), 0) + 1
for bot in submitted_bots:
if bot.name not in owner_by_bot:
active_by_owner[owner] = active_by_owner.get(owner, 0) + 1
if active_by_owner.get(owner, 0) > 5:
raise ValidationError(f"owner `{owner}` exceeds the five active bot slot limit")

Expand All @@ -172,10 +180,28 @@ def generated_catalog(bots: list[Bot], root: Path, owner: str) -> tuple[dict[str
for bot in bots:
bot_owner = owner_by_bot.get(bot.name, owner)
records.setdefault(bot_owner, []).append(bot.name)
owners = {record["ownerId"]: set(record.get("bots", [])) for record in existing_owners.get("owners", [])}
owners = {
record["ownerId"]: {
"accounts": record.get("accounts", []),
"bots": set(record.get("bots", [])),
}
for record in existing_owners.get("owners", [])
}
for owner_id, names in records.items():
owners.setdefault(owner_id, set()).update(names)
owner_data = {"schemaVersion": 1, "owners": [{"ownerId": owner_id, "accounts": [owner_id], "bots": sorted(names), "activeSlots": len(names)} for owner_id, names in sorted(owners.items())]}
owner_record = owners.setdefault(owner_id, {"accounts": [owner_id], "bots": set()})
owner_record["bots"].update(names)
owner_data = {
"schemaVersion": 1,
"owners": [
{
"ownerId": owner_id,
"accounts": owner_record["accounts"],
"bots": sorted(owner_record["bots"]),
"activeSlots": len(owner_record["bots"]),
}
for owner_id, owner_record in sorted(owners.items())
],
}
today = datetime.now(UTC).date().isoformat()
current_by_name = {bot.name: bot for bot in bots}
history = []
Expand Down
40 changes: 38 additions & 2 deletions tests/test_validate_bot.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,8 +21,20 @@ def setUp(self) -> None:
def tearDown(self) -> None:
self.temporary_directory.cleanup()

def run_validator(self, *arguments: str) -> subprocess.CompletedProcess[str]:
return subprocess.run([sys.executable, str(VALIDATOR), "--root", str(self.root), "--owner", "flemming-n-larsen", *arguments], text=True, capture_output=True, check=False)
def run_validator(self, *arguments: str, owner: str = "flemming-n-larsen") -> subprocess.CompletedProcess[str]:
return subprocess.run([sys.executable, str(VALIDATOR), "--root", str(self.root), "--owner", owner, *arguments], text=True, capture_output=True, check=False)

def add_bot(self, name: str) -> None:
source = self.root / "bots" / "python" / "Orbit"
destination = self.root / "bots" / "python" / name
shutil.copytree(source, destination)
(destination / "Orbit.sh").rename(destination / f"{name}.sh")
(destination / "Orbit.cmd").rename(destination / f"{name}.cmd")
config_path = destination / "Orbit.json"
config = json.loads(config_path.read_text(encoding="utf-8"))
config["name"] = name
config_path.unlink()
(destination / f"{name}.json").write_text(json.dumps(config), encoding="utf-8")

def test_valid_submission_generates_an_active_catalog_entry(self) -> None:
result = self.run_validator("--smoke", "--generate")
Expand Down Expand Up @@ -58,6 +70,30 @@ def test_version_increase_supersedes_the_previous_catalog_entry(self) -> None:
catalog = json.loads((self.root / "bots" / "index.json").read_text(encoding="utf-8"))
self.assertEqual(["superseded", "active"], [entry["status"] for entry in catalog["bots"]])

def test_new_bot_from_another_owner_ignores_unchanged_catalog_entries(self) -> None:
self.assertEqual(0, self.run_validator("--generate").returncode)
self.add_bot("Nova")
result = self.run_validator("--generate", owner="alice")
self.assertEqual(0, result.returncode, result.stderr)
catalog = json.loads((self.root / "bots" / "index.json").read_text(encoding="utf-8"))
nova = next(entry for entry in catalog["bots"] if entry["name"] == "Nova")
self.assertEqual("alice", nova["owner"])

def test_registered_secondary_account_can_update_and_is_preserved(self) -> None:
owners_path = self.root / "bots" / "owners.json"
owners = json.loads(owners_path.read_text(encoding="utf-8"))
owners["owners"][0]["ownerId"] = "primary"
owners["owners"][0]["accounts"] = ["primary", "secondary"]
owners_path.write_text(json.dumps(owners), encoding="utf-8")
config_path = self.root / "bots" / "python" / "Orbit" / "Orbit.json"
config = json.loads(config_path.read_text(encoding="utf-8"))
config["version"] = "1.0.3"
config_path.write_text(json.dumps(config), encoding="utf-8")
result = self.run_validator("--generate", owner="secondary")
self.assertEqual(0, result.returncode, result.stderr)
regenerated_owners = json.loads(owners_path.read_text(encoding="utf-8"))
self.assertEqual(["primary", "secondary"], regenerated_owners["owners"][0]["accounts"])


if __name__ == "__main__":
unittest.main()
Loading