231 lines
7.1 KiB
Bash
Executable File
231 lines
7.1 KiB
Bash
Executable File
#!/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
|