refactor(examples): improve code quality and safety across examples (#12)

- Add HTTP server read/write/idle timeouts to go-webservice to prevent slowloris attacks
- Log exception details in python-cli token validation instead of silently swallowing
- Extract duplicated scopes list into SCOPES constant in python-cli
- Simplify python-m2m response reading from streaming chunk iterator to simple get
- Extract triplicated symlink/ownership guard into validate_cache_file helper in bash-cli

Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: appleboy

bash-cli/main.sh

@@ -224,15 +224,19 @@ discover_endpoints() {
 
 # --- Token Cache ---
 
-load_cached_token() {
-  [ -f "$TOKEN_CACHE_FILE" ] || return 1
-
-  # Refuse to operate on a symlink or a file not owned by the current user
-  # to avoid following attacker-controlled links to credential files.
+# Refuse to operate on a symlink or a file not owned by the current user
+# to avoid following attacker-controlled links to credential files.
+validate_cache_file() {
   if [ -L "$TOKEN_CACHE_FILE" ] || [ ! -O "$TOKEN_CACHE_FILE" ]; then
     echo "Warning: Refusing to use token cache file that is a symlink or not owned by the current user: $TOKEN_CACHE_FILE" >&2
     return 1
   fi
+  return 0
+}
+
+load_cached_token() {
+  [ -f "$TOKEN_CACHE_FILE" ] || return 1
+  validate_cache_file || return 1
 
   chmod 600 "$TOKEN_CACHE_FILE" 2>/dev/null || true
 
@@ -298,10 +302,7 @@ save_cached_token() {
   # If the cache file already exists, refuse to operate on a symlink or
   # a file not owned by the current user to prevent credential clobbering.
   if [ -e "$TOKEN_CACHE_FILE" ]; then
-    if [ -L "$TOKEN_CACHE_FILE" ] || [ ! -O "$TOKEN_CACHE_FILE" ]; then
-      echo "Warning: Refusing to write token cache file that is a symlink or not owned by the current user: $TOKEN_CACHE_FILE" >&2
-      return 1
-    fi
+    validate_cache_file || return 1
   fi
 
   # Treat missing or corrupted cache as empty; fall back to {}
@@ -331,11 +332,7 @@ save_cached_token() {
 delete_cached_token() {
   [ -f "$TOKEN_CACHE_FILE" ] || return 0
 
-  # Refuse to operate on a symlink or a file not owned by the current user
-  if [ -L "$TOKEN_CACHE_FILE" ] || [ ! -O "$TOKEN_CACHE_FILE" ]; then
-    echo "Warning: Refusing to delete token cache file that is a symlink or not owned by the current user: $TOKEN_CACHE_FILE" >&2
-    return 0
-  fi
+  validate_cache_file || return 0
 
   local tmp
   tmp=$(mktemp "${TOKEN_CACHE_FILE}.XXXXXX")

go-webservice/main.go

@@ -21,6 +21,7 @@ import (
 	"log"
 	"net/http"
 	"os"
+	"time"
 
 	"github.com/go-authgate/sdk-go/discovery"
 	"github.com/go-authgate/sdk-go/middleware"
@@ -70,8 +71,15 @@ func main() {
 	mux.Handle("/api/data", authWithScope(http.HandlerFunc(dataHandler)))
 	mux.HandleFunc("/health", healthHandler)
 
+	srv := &http.Server{
+		Addr:         ":8080",
+		Handler:      mux,
+		ReadTimeout:  15 * time.Second,
+		WriteTimeout: 15 * time.Second,
+		IdleTimeout:  60 * time.Second,
+	}
 	log.Println("Listening on :8080")
-	log.Fatal(http.ListenAndServe(":8080", mux))
+	log.Fatal(srv.ListenAndServe())
 }
 
 func profileHandler(w http.ResponseWriter, r *http.Request) {

python-cli/main.py

@@ -20,6 +20,8 @@
 import authgate
 from authgate.credstore import default_token_secure_store
 
+SCOPES = ["profile", "email"]
+
 
 def mask_token(s: str) -> str:
     if len(s) <= 8:
@@ -76,20 +78,20 @@ def main():
     client, token = authgate.authenticate(
         authgate_url,
         client_id,
-        scopes=["profile", "email"],
+        scopes=SCOPES,
     )
 
     # If the cached token is revoked/expired server-side, clear it and re-authenticate.
     try:
         info = client.userinfo(token.access_token)
-    except Exception:
-        print("Cached token is invalid, re-authenticating...")
+    except Exception as e:
+        print(f"Cached token is invalid ({e}), re-authenticating...")
         store = default_token_secure_store("authgate", ".authgate-tokens.json")
         store.delete(client_id)
         client, token = authgate.authenticate(
             authgate_url,
             client_id,
-            scopes=["profile", "email"],
+            scopes=SCOPES,
         )
         info = None
 

python-m2m/main.py

@@ -53,25 +53,13 @@ def main():
     # 4. Use the auto-authenticated HTTP client
     auth = BearerAuth(ts)
     with httpx.Client(auth=auth) as http:
-        with http.stream("GET", f"{authgate_url}/oauth/userinfo") as resp:
-            status_code = resp.status_code
-            body_bytes = bytearray()
-            for chunk in resp.iter_bytes():
-                if not chunk:
-                    continue
-                remaining = (MAX_BODY_SIZE + 1) - len(body_bytes)
-                if remaining <= 0:
-                    break
-                if len(chunk) > remaining:
-                    body_bytes.extend(chunk[:remaining])
-                    break
-                body_bytes.extend(chunk)
-            body = bytes(body_bytes)
+        resp = http.get(f"{authgate_url}/oauth/userinfo")
+        body = resp.content
 
     truncated = len(body) > MAX_BODY_SIZE
     if truncated:
         body = body[:MAX_BODY_SIZE]
-    print(f"Status: {status_code}")
+    print(f"Status: {resp.status_code}")
     print(f"Body: {body.decode(errors='replace')}")
     if truncated:
         print("(response body truncated to 1 MB)")