Report token scan results in tables
This commit is contained in:
@@ -425,45 +425,98 @@ def sync_tokens_from_repo(config: WorkspaceConfig, repo_path: Path) -> dict:
|
||||
if repo_root is None:
|
||||
return {
|
||||
"repo_root": None,
|
||||
"remotes": 0,
|
||||
"urls": 0,
|
||||
"scanned": 0,
|
||||
"auth_urls": 0,
|
||||
"plain_urls": 0,
|
||||
"unsupported_urls": 0,
|
||||
"found": 0,
|
||||
"added": 0,
|
||||
"servers": 0,
|
||||
"remote_rows": [],
|
||||
"changed": False,
|
||||
}
|
||||
|
||||
found_credentials: list[tuple[dict, str, str]] = []
|
||||
for remote_url in remote_urls(repo_root):
|
||||
found_credentials: list[tuple[int, dict, str, str]] = []
|
||||
remote_entries = remote_name_urls(repo_root)
|
||||
remote_names = {remote_name for remote_name, _ in remote_entries}
|
||||
touched_servers: set[str] = set()
|
||||
remote_rows: list[dict[str, str]] = []
|
||||
auth_urls = 0
|
||||
plain_urls = 0
|
||||
unsupported_urls = 0
|
||||
for remote_name, remote_url in remote_entries:
|
||||
server_info = server_info_from_url(config, remote_url)
|
||||
credentials = remote_credentials(remote_url)
|
||||
if server_info is not None and credentials is not None:
|
||||
user_name, token_value = credentials
|
||||
found_credentials.append((server_info, user_name, token_value))
|
||||
if server_info is None:
|
||||
unsupported_urls += 1
|
||||
remote_rows.append(
|
||||
{
|
||||
"remote": remote_name,
|
||||
"endpoint": "",
|
||||
"type": "unsupported",
|
||||
"url_kind": "unsupported",
|
||||
"user": "",
|
||||
"result": "ignored",
|
||||
}
|
||||
)
|
||||
continue
|
||||
|
||||
if not found_credentials:
|
||||
return {
|
||||
"repo_root": repo_root,
|
||||
"scanned": 0,
|
||||
"added": 0,
|
||||
"servers": 0,
|
||||
"changed": False,
|
||||
}
|
||||
touched_servers.add(server_info["endpoint"])
|
||||
credentials = remote_credentials(remote_url)
|
||||
if credentials is None:
|
||||
plain_urls += 1
|
||||
remote_rows.append(
|
||||
{
|
||||
"remote": remote_name,
|
||||
"endpoint": server_info["endpoint"],
|
||||
"type": server_info["type"],
|
||||
"url_kind": "plain",
|
||||
"user": "",
|
||||
"result": "no_credentials",
|
||||
}
|
||||
)
|
||||
continue
|
||||
|
||||
auth_urls += 1
|
||||
user_name, token_value = credentials
|
||||
row_index = len(remote_rows)
|
||||
remote_rows.append(
|
||||
{
|
||||
"remote": remote_name,
|
||||
"endpoint": server_info["endpoint"],
|
||||
"type": server_info["type"],
|
||||
"url_kind": "auth",
|
||||
"user": user_name,
|
||||
"result": "found",
|
||||
}
|
||||
)
|
||||
found_credentials.append((row_index, server_info, user_name, token_value))
|
||||
|
||||
token_data = load_token_store(config)
|
||||
added = 0
|
||||
touched_servers: set[str] = set()
|
||||
for server_info, user_name, token_value in found_credentials:
|
||||
touched_servers.add(server_info["endpoint"])
|
||||
for row_index, server_info, user_name, token_value in found_credentials:
|
||||
if register_token(token_data, server_info, user_name, token_value):
|
||||
added += 1
|
||||
remote_rows[row_index]["result"] = "added"
|
||||
else:
|
||||
remote_rows[row_index]["result"] = "existing"
|
||||
|
||||
if added or not config.token_path.exists():
|
||||
if found_credentials and (added or not config.token_path.exists()):
|
||||
write_token_store(config, token_data)
|
||||
|
||||
return {
|
||||
"repo_root": repo_root,
|
||||
"scanned": len(found_credentials),
|
||||
"remotes": len(remote_names),
|
||||
"urls": len(remote_entries),
|
||||
"scanned": auth_urls + plain_urls,
|
||||
"auth_urls": auth_urls,
|
||||
"plain_urls": plain_urls,
|
||||
"unsupported_urls": unsupported_urls,
|
||||
"found": len(found_credentials),
|
||||
"added": added,
|
||||
"servers": len(touched_servers),
|
||||
"remote_rows": remote_rows,
|
||||
"changed": added > 0,
|
||||
}
|
||||
|
||||
@@ -656,11 +709,25 @@ def resolve_repo_argument(repo_arg: str | None) -> Path:
|
||||
def run_tokens_scan(config: WorkspaceConfig, args: argparse.Namespace) -> None:
|
||||
report = sync_tokens_from_repo(config, resolve_repo_argument(args.repo))
|
||||
repo_root = report["repo_root"]
|
||||
print(f"repo_root\t{repo_root or ''}")
|
||||
print(f"scanned\t{report['scanned']}")
|
||||
print(f"added\t{report['added']}")
|
||||
print(f"servers\t{report['servers']}")
|
||||
print(f"token_path\t{config.token_path}")
|
||||
print("summary")
|
||||
print(
|
||||
"repo_root\tremotes\turls\tscanned\tauth_urls\tplain_urls\t"
|
||||
"unsupported_urls\tfound\tadded\tservers\ttoken_path"
|
||||
)
|
||||
print(
|
||||
f"{repo_root or ''}\t{report['remotes']}\t{report['urls']}\t"
|
||||
f"{report['scanned']}\t{report['auth_urls']}\t{report['plain_urls']}\t"
|
||||
f"{report['unsupported_urls']}\t{report['found']}\t{report['added']}\t"
|
||||
f"{report['servers']}\t{config.token_path}"
|
||||
)
|
||||
print()
|
||||
print("remotes")
|
||||
print("remote\tendpoint\ttype\turl_kind\tuser\tresult")
|
||||
for row in report["remote_rows"]:
|
||||
print(
|
||||
f"{row['remote']}\t{row['endpoint']}\t{row['type']}\t"
|
||||
f"{row['url_kind']}\t{row['user']}\t{row['result']}"
|
||||
)
|
||||
|
||||
|
||||
def run_tokens_read(config: WorkspaceConfig, args: argparse.Namespace) -> None:
|
||||
|
||||
Reference in New Issue
Block a user