add: devops scripts and configs
This commit is contained in:
“Naeel”
Executable
+230
@@ -0,0 +1,230 @@
|
||||
#!/usr/bin/env bash
|
||||
set -euo pipefail
|
||||
|
||||
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
|
||||
DEFAULT_BASE_DIR="$(cd "${SCRIPT_DIR}/.." && pwd)"
|
||||
BASE_DIR="${BASE_DIR:-${DEFAULT_BASE_DIR}}"
|
||||
CLIENT_ID="${CLIENT_ID:-account-console}"
|
||||
TOKEN_URL="${TOKEN_URL:-https://keycloak.nubes.ru/realms/cloud/protocol/openid-connect/token}"
|
||||
THRESHOLD_MIN="${THRESHOLD_MIN:-10}"
|
||||
FORCE_REFRESH="${FORCE_REFRESH:-0}"
|
||||
|
||||
usage() {
|
||||
echo "Usage:" >&2
|
||||
echo " $0 save [json_file|-]" >&2
|
||||
echo " $0 refresh" >&2
|
||||
echo " $0 current" >&2
|
||||
echo "" >&2
|
||||
echo "Env vars:" >&2
|
||||
echo " BASE_DIR=${DEFAULT_BASE_DIR}" >&2
|
||||
echo " THRESHOLD_MIN=10" >&2
|
||||
echo " FORCE_REFRESH=1" >&2
|
||||
echo " TOKEN_URL=https://keycloak.nubes.ru/realms/cloud/protocol/openid-connect/token" >&2
|
||||
echo " CLIENT_ID=account-console" >&2
|
||||
}
|
||||
|
||||
cmd="${1:-}"
|
||||
|
||||
case "$cmd" in
|
||||
save)
|
||||
json_source="${2:--}"
|
||||
if [[ "$json_source" != "-" && ! -f "$json_source" ]]; then
|
||||
echo "Error: JSON file not found: $json_source" >&2
|
||||
exit 2
|
||||
fi
|
||||
if [[ "$json_source" == "-" ]]; then
|
||||
cat
|
||||
else
|
||||
cat "$json_source"
|
||||
fi | BASE_DIR="$BASE_DIR" python3 - <<'PY'
|
||||
import base64
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from datetime import datetime, timezone, timedelta
|
||||
|
||||
base_dir = os.environ.get("BASE_DIR")
|
||||
raw = sys.stdin.read().strip()
|
||||
if not raw:
|
||||
print("Error: empty JSON input", file=sys.stderr)
|
||||
sys.exit(2)
|
||||
|
||||
data = json.loads(raw)
|
||||
access = data.get("access_token")
|
||||
refresh = data.get("refresh_token")
|
||||
if not access:
|
||||
print("Error: access_token missing", file=sys.stderr)
|
||||
sys.exit(2)
|
||||
|
||||
def decode_exp(token):
|
||||
payload = token.split(".")[1]
|
||||
padding = "=" * (-len(payload) % 4)
|
||||
decoded = base64.urlsafe_b64decode(payload + padding)
|
||||
payload_json = json.loads(decoded)
|
||||
exp = payload_json.get("exp")
|
||||
iat = payload_json.get("iat")
|
||||
if not exp:
|
||||
raise ValueError("exp missing in token payload")
|
||||
return exp, iat
|
||||
|
||||
|
||||
def save_token(token, suffix):
|
||||
exp, iat = decode_exp(token)
|
||||
fname = datetime.fromtimestamp(exp, tz=timezone(timedelta(hours=4))).strftime("%H-%M-%S")
|
||||
path = os.path.join(base_dir, f"{fname}.{suffix}")
|
||||
with open(path, "w", encoding="utf-8") as f:
|
||||
f.write(token)
|
||||
return path, exp, iat
|
||||
|
||||
access_path, access_exp, access_iat = save_token(access, "token")
|
||||
print(access_path)
|
||||
if refresh:
|
||||
refresh_path, refresh_exp, refresh_iat = save_token(refresh, "refresh")
|
||||
print(refresh_path)
|
||||
PY
|
||||
;;
|
||||
refresh)
|
||||
BASE_DIR="$BASE_DIR" TOKEN_URL="$TOKEN_URL" CLIENT_ID="$CLIENT_ID" THRESHOLD_MIN="$THRESHOLD_MIN" FORCE_REFRESH="$FORCE_REFRESH" python3 - <<'PY'
|
||||
import base64
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from datetime import datetime, timezone, timedelta
|
||||
from urllib import request, parse
|
||||
|
||||
base_dir = os.environ.get("BASE_DIR")
|
||||
url = os.environ.get("TOKEN_URL")
|
||||
client_id = os.environ.get("CLIENT_ID")
|
||||
threshold_min = int(os.environ.get("THRESHOLD_MIN", "10"))
|
||||
force_refresh = os.environ.get("FORCE_REFRESH", "0") == "1"
|
||||
|
||||
|
||||
def decode_exp(token):
|
||||
payload = token.split(".")[1]
|
||||
padding = "=" * (-len(payload) % 4)
|
||||
decoded = base64.urlsafe_b64decode(payload + padding)
|
||||
payload_json = json.loads(decoded)
|
||||
exp = payload_json.get("exp")
|
||||
iat = payload_json.get("iat")
|
||||
if not exp:
|
||||
raise ValueError("exp missing in token payload")
|
||||
return exp, iat
|
||||
|
||||
|
||||
def newest_file(pattern):
|
||||
import glob
|
||||
files = glob.glob(pattern)
|
||||
if not files:
|
||||
return None
|
||||
files.sort(key=lambda p: os.path.getmtime(p), reverse=True)
|
||||
return files[0]
|
||||
|
||||
access_path = newest_file(os.path.join(base_dir, "*.token"))
|
||||
refresh_path = newest_file(os.path.join(base_dir, "*.refresh"))
|
||||
if not refresh_path:
|
||||
print("Error: no refresh token file found", file=sys.stderr)
|
||||
sys.exit(2)
|
||||
|
||||
with open(refresh_path, "r", encoding="utf-8") as f:
|
||||
refresh_token = f.read().strip()
|
||||
|
||||
refresh_exp, refresh_iat = decode_exp(refresh_token)
|
||||
now = int(datetime.now(tz=timezone.utc).timestamp())
|
||||
if refresh_exp <= now:
|
||||
print("Error: refresh token expired", file=sys.stderr)
|
||||
sys.exit(2)
|
||||
|
||||
if access_path:
|
||||
with open(access_path, "r", encoding="utf-8") as f:
|
||||
access_token = f.read().strip()
|
||||
access_exp, access_iat = decode_exp(access_token)
|
||||
if not force_refresh and access_exp - now > threshold_min * 60:
|
||||
fname = datetime.fromtimestamp(access_exp, tz=timezone(timedelta(hours=4))).strftime("%H-%M-%S")
|
||||
print(access_path)
|
||||
print(f"Access token valid until {fname} GMT+4; no refresh needed")
|
||||
sys.exit(0)
|
||||
|
||||
post_data = parse.urlencode({
|
||||
"grant_type": "refresh_token",
|
||||
"client_id": client_id,
|
||||
"refresh_token": refresh_token,
|
||||
}).encode("utf-8")
|
||||
req = request.Request(url, data=post_data, method="POST")
|
||||
req.add_header("Content-Type", "application/x-www-form-urlencoded")
|
||||
|
||||
try:
|
||||
with request.urlopen(req, timeout=30) as resp:
|
||||
raw = resp.read().decode("utf-8")
|
||||
except Exception as exc:
|
||||
print("HTTP error:", exc, file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
try:
|
||||
data = json.loads(raw)
|
||||
except Exception:
|
||||
print("Failed to parse response as JSON", file=sys.stderr)
|
||||
print(raw, file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
if "access_token" not in data:
|
||||
print("Refresh failed:", data.get("error"), "-", data.get("error_description"), file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
access_token = data.get("access_token")
|
||||
new_refresh_token = data.get("refresh_token")
|
||||
|
||||
access_exp, access_iat = decode_exp(access_token)
|
||||
access_fname = datetime.fromtimestamp(access_exp, tz=timezone(timedelta(hours=4))).strftime("%H-%M-%S")
|
||||
access_out = os.path.join(base_dir, f"{access_fname}.token")
|
||||
with open(access_out, "w", encoding="utf-8") as f:
|
||||
f.write(access_token)
|
||||
print(access_out)
|
||||
|
||||
if new_refresh_token:
|
||||
refresh_exp, refresh_iat = decode_exp(new_refresh_token)
|
||||
refresh_fname = datetime.fromtimestamp(refresh_exp, tz=timezone(timedelta(hours=4))).strftime("%H-%M-%S")
|
||||
refresh_out = os.path.join(base_dir, f"{refresh_fname}.refresh")
|
||||
with open(refresh_out, "w", encoding="utf-8") as f:
|
||||
f.write(new_refresh_token)
|
||||
print(refresh_out)
|
||||
PY
|
||||
;;
|
||||
current)
|
||||
BASE_DIR="$BASE_DIR" python3 - <<'PY'
|
||||
import base64
|
||||
import json
|
||||
import os
|
||||
from datetime import datetime, timezone, timedelta
|
||||
|
||||
base_dir = os.environ.get("BASE_DIR")
|
||||
|
||||
def decode_exp(token):
|
||||
payload = token.split(".")[1]
|
||||
padding = "=" * (-len(payload) % 4)
|
||||
decoded = base64.urlsafe_b64decode(payload + padding)
|
||||
payload_json = json.loads(decoded)
|
||||
exp = payload_json.get("exp")
|
||||
iat = payload_json.get("iat")
|
||||
if not exp:
|
||||
raise ValueError("exp missing in token payload")
|
||||
return exp, iat
|
||||
|
||||
paths = [p for p in os.listdir(base_dir) if p.endswith(".token")]
|
||||
if not paths:
|
||||
print("No access token files found")
|
||||
raise SystemExit(2)
|
||||
paths.sort(key=lambda p: os.path.getmtime(os.path.join(base_dir, p)), reverse=True)
|
||||
path = os.path.join(base_dir, paths[0])
|
||||
with open(path, "r", encoding="utf-8") as f:
|
||||
token = f.read().strip()
|
||||
exp, iat = decode_exp(token)
|
||||
print(path)
|
||||
print("iat", datetime.fromtimestamp(iat, tz=timezone(timedelta(hours=4))).strftime("%Y-%m-%d %H:%M:%S GMT+4"))
|
||||
print("exp", datetime.fromtimestamp(exp, tz=timezone(timedelta(hours=4))).strftime("%Y-%m-%d %H:%M:%S GMT+4"))
|
||||
PY
|
||||
;;
|
||||
*)
|
||||
usage
|
||||
exit 2
|
||||
;;
|
||||
esac
|
||||
Reference in New Issue
Block a user