forked from k-space/doorboy-proxy
e6fc5cb85f
/allowed returns keep_open_until from the newest approved hold in doorlog; /longpoll skips hold events to avoid spurious open pulses. Fixes: assert->raise for SECRET check, text() on 403, remove dead /logs code, flatten auth decorator, by_slackid None fallback, load kube config once, guard missing slack command, backoff on PyMongoError, mongo->mongosh.
60 lines
1.6 KiB
Python
60 lines
1.6 KiB
Python
import os
|
|
from typing import List, Optional, Tuple
|
|
|
|
from kubernetes import client, config
|
|
|
|
OIDC_USERS_NAMESPACE = os.environ["OIDC_USERS_NAMESPACE"]
|
|
|
|
_config_loaded = False
|
|
|
|
|
|
def _ensure_config():
|
|
"""Load in-cluster Kubernetes config exactly once (lazy, cached)."""
|
|
global _config_loaded
|
|
if not _config_loaded:
|
|
config.load_incluster_config()
|
|
_config_loaded = True
|
|
|
|
|
|
def groupsToFullName(groups) -> List[str]:
|
|
fullName: List[str] = []
|
|
|
|
for group in groups:
|
|
fullName.append(
|
|
group.get("prefix", "") + ":" + group.get("name", "")
|
|
)
|
|
|
|
return fullName
|
|
|
|
|
|
def _get_users() -> list:
|
|
"""Return all OIDC user items from the Kubernetes API."""
|
|
_ensure_config()
|
|
api_instance = client.CustomObjectsApi()
|
|
ret = api_instance.list_namespaced_custom_object(
|
|
"codemowers.cloud", "v1beta1", OIDC_USERS_NAMESPACE, "oidcusers"
|
|
)
|
|
return ret["items"]
|
|
|
|
|
|
def users_with_group(requiredGroup: str) -> List[str]:
|
|
users: List[str] = []
|
|
|
|
for item in _get_users():
|
|
for group in groupsToFullName(item.get("status", {}).get("groups", [])):
|
|
if group == requiredGroup:
|
|
users.append(item["metadata"]["name"])
|
|
continue
|
|
|
|
print(f"INFO: {len(users)} users in group {requiredGroup}")
|
|
return users
|
|
|
|
|
|
# -> (groups[], username)
|
|
def by_slackid(slack_id: str) -> Tuple[List[str], Optional[str]]:
|
|
for item in _get_users():
|
|
if slack_id == item.get("status", {}).get("slackId", None):
|
|
return groupsToFullName(item.get("status", {}).get("groups", [])), item.get("metadata", {}).get("name", "")
|
|
|
|
return [], None
|