tf_provider/TOOLS/scripts/00_token_manager.sh

#!/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