From b8234f9fa6fc5c962ab4886cbf371a1d8cd9d513 Mon Sep 17 00:00:00 2001 From: Jan Malte Behrens Date: Tue, 19 May 2026 16:20:16 +0200 Subject: [PATCH 01/17] feat: Switching auth jwt parsing method to match keycloak --- auth/auth.go | 249 ++++++++++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 248 insertions(+), 1 deletion(-) diff --git a/auth/auth.go b/auth/auth.go index 0941353e8..ff9ad0f58 100644 --- a/auth/auth.go +++ b/auth/auth.go @@ -5,9 +5,12 @@ package auth import ( "context" "encoding/json" + "encoding/base64" "errors" "fmt" + "math/big" "net/http" + "crypto/rsa" "slices" "strconv" "strings" @@ -33,6 +36,12 @@ var ( envAuthTokenFile = environment.NewVariable("AUTH_TOKEN_KEY_FILE", "/run/secrets/auth_token_key", "Key to sign the JWT auth tocken.") envAuthCookieFile = environment.NewVariable("AUTH_COOKIE_KEY_FILE", "/run/secrets/auth_cookie_key", "Key to sign the JWT auth cookie.") + + envIssuerURL = environment.NewVariable("OIDC_ISSUER_URL", "http://localhost:8080/realms/openslides", "URL of keycloak server") + envIssuerURLDocker = environment.NewVariable("OIDC_ISSUER_URL_DOCKER", "http://keycloak-server:8080/realms/openslides", "Dockerized URL of keycloak server") + envClientID = environment.NewVariable("OIDC_CLIENT_ID", "proxy-client", "Keycloak client name") + envClientSecret = environment.NewVariable("OIDC_CLIENT_SECRET", "proxy-secret", "Keycloak client secret") + envSecret = environment.NewVariable("OIDC_SECRET", "qvAcTGWBIGg7aWKCKRyUsTf33jK3lsmK", "Keycloak secret") ) // pruneTime defines how long a topic id will be valid. This should be higher @@ -41,7 +50,7 @@ const pruneTime = 15 * time.Minute const ( cookieName = "refreshId" - authHeader = "Authentication" + authHeader = "Authorization" authPath = "/internal/auth/authenticate" ) @@ -65,6 +74,11 @@ type Auth struct { tokenKey string cookieKey string + + issuerURLDocker string + + keys map[string]*rsa.PublicKey + expiresAt time.Time } // New initializes the Auth object. @@ -91,12 +105,19 @@ func New(lookup environment.Environmenter, messageBus LogoutEventer) (*Auth, fun return nil, nil, fmt.Errorf("reading cookie token: %w", err) } + issuerURLDocker := envIssuerURLDocker.Value(lookup) + if issuerURLDocker == "" { + issuerURLDocker = envIssuerURL.Value(lookup) + } + a := &Auth{ fake: fake, logedoutSessions: topic.New[string](), authServiceURL: url, tokenKey: authToken, cookieKey: cookieToken, + issuerURLDocker: issuerURLDocker, + keys: make(map[string]*rsa.PublicKey), } // Make sure the topic is not empty @@ -114,6 +135,7 @@ func New(lookup environment.Environmenter, messageBus LogoutEventer) (*Auth, fun return a, background, nil } + // Authenticate uses the headers from the given request to get the user id. The // returned context will be cancled, if the session is revoked. func (a *Auth) Authenticate(w http.ResponseWriter, r *http.Request) (context.Context, error) { @@ -123,6 +145,231 @@ func (a *Auth) Authenticate(w http.ResponseWriter, r *http.Request) (context.Con ctx := r.Context() + p := new(payloadKeycloak) + if err := a.loadTokenKeycloak(w, r, p); err != nil { + fmt.Println("reading token: %w", err) + return nil, fmt.Errorf("reading token: %w", err) + } + + fmt.Println("We are here babeeey") + if p.KeycloakID == "" { + return a.AuthenticatedContext(ctx, 0), nil + } + + cid, sessionIDs := a.logedoutSessions.ReceiveAll() + if slices.Contains(sessionIDs, p.SessionID) { + return nil, &authError{"invalid session", nil} + } + + // Get OS User Id linked to Keycloak ID + userID := 1 + + ctx, cancelCtx := context.WithCancel(a.AuthenticatedContext(ctx, userID)) + + go func() { + defer cancelCtx() + + var sessionIDs []string + var err error + for { + cid, sessionIDs, err = a.logedoutSessions.ReceiveSince(ctx, cid) + if err != nil { + return + } + + if slices.Contains(sessionIDs, p.SessionID) { + return + } + } + }() + + return ctx, nil +} + +// loadToken loads and validates the token. If the token is expired, it tries +// to renew it and write the new token in the responsewriter. +func (a *Auth) loadTokenKeycloak(w http.ResponseWriter, r *http.Request, payload jwt.Claims) error { + header := r.Header.Get(authHeader) + encodedToken := strings.TrimPrefix(header, "Bearer: ") + + if header == encodedToken { + // No token. Handle the request as public access requst. + return nil + } + + _, err := a.validateToken(r.Context(), encodedToken) + if err != nil { + // OIDC validation failed - return error (no cookie means no legacy fallback) + return authError{msg: "Invalid OIDC token", wrapped: err} + } + + _, err = jwt.ParseWithClaims(encodedToken, payload, func(token *jwt.Token) (interface{}, error) { + return []byte(a.tokenKey), nil + }) + + if err != nil { + var invalid *jwt.ValidationError + if errors.As(err, &invalid) { + fmt.Println(err) + return a.handleInvalidToken(r.Context(), invalid, w, encodedToken, "") + } + } + + return nil +} + +func (a *Auth) validateToken(ctx context.Context, tokenString string) (*payloadKeycloak, error) { + // 1. Parse token without validation to get kid + token, _, err := new(jwt.Parser).ParseUnverified(tokenString, &payloadKeycloak{}) + if err != nil { + return nil, fmt.Errorf("parsing token: %w", err) + } + + kid, ok := token.Header["kid"].(string) + if !ok { + return nil, errors.New("missing kid in token header") + } + + // 2. Get public key from JWKS (cached) + key, err := a.getKey(ctx, kid) + if err != nil { + return nil, fmt.Errorf("getting key: %w", err) + } + + // 3. Validate token with public key + claims := &payloadKeycloak{} + token, err = jwt.ParseWithClaims(tokenString, claims, func(t *jwt.Token) (interface{}, error) { + if _, ok := t.Method.(*jwt.SigningMethodRSA); !ok { + return nil, fmt.Errorf("unexpected signing method: %v", t.Header["alg"]) + } + return key, nil + }) + if err != nil { + return nil, fmt.Errorf("validating token: %w", err) + } + + if !token.Valid { + return nil, errors.New("invalid token") + } + + // 4. Validate issuer + //if claims.Issuer != a.issuerURLDocker { + // return nil, fmt.Errorf("invalid issuer: got %s, want %s", claims.Issuer, a.issuerURLDocker) + //} + + return claims, nil +} + +type payloadKeycloak struct { + jwt.RegisteredClaims + KeycloakID string `json:"sub"` + SessionID string `json:"sid"` // Keycloak session ID + Email string `json:"email"` + Username string `json:"preferred_username"` + ClientName string `json:"azp"` +} + +// getKey returns the RSA public key for the given kid, fetching from JWKS if needed +func (a *Auth) getKey(ctx context.Context, kid string) (*rsa.PublicKey, error) { + if key, ok := a.keys[kid]; ok && time.Now().Before(a.expiresAt) { + return key, nil + } + + // Fetch JWKS + return a.fetchJWKS(ctx, kid) +} + +// fetchJWKS fetches the JWKS from the issuer and caches the keys +func (a *Auth) fetchJWKS(ctx context.Context, kid string) (*rsa.PublicKey, error) { + // Double-check after acquiring write lock + if key, ok := a.keys[kid]; ok && time.Now().Before(a.expiresAt) { + return key, nil + } + + req, err := http.NewRequestWithContext(ctx, "GET", a.issuerURLDocker + "/protocol/openid-connect/certs", nil) + if err != nil { + return nil, fmt.Errorf("creating JWKS request: %w", err) + } + + resp, err := http.DefaultClient.Do(req) + if err != nil { + return nil, fmt.Errorf("fetching JWKS: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("JWKS request failed: %s", resp.Status) + } + + var jwks struct { + Keys []struct { + Kid string `json:"kid"` + Kty string `json:"kty"` + Alg string `json:"alg"` + N string `json:"n"` + E string `json:"e"` + } `json:"keys"` + } + + if err := json.NewDecoder(resp.Body).Decode(&jwks); err != nil { + return nil, fmt.Errorf("decoding JWKS: %w", err) + } + + // Parse and cache all keys + a.keys = make(map[string]*rsa.PublicKey) + for _, k := range jwks.Keys { + if k.Kty != "RSA" { + continue + } + key, err := parseRSAPublicKey(k.N, k.E) + if err != nil { + continue + } + a.keys[k.Kid] = key + } + + // Cache for 1 hour + a.expiresAt = time.Now().Add(time.Hour) + + key, ok := a.keys[kid] + if !ok { + return nil, fmt.Errorf("key %s not found in JWKS", kid) + } + return key, nil +} + +// parseRSAPublicKey parses RSA public key from JWKS n and e values +func parseRSAPublicKey(nStr, eStr string) (*rsa.PublicKey, error) { + nBytes, err := base64.RawURLEncoding.DecodeString(nStr) + if err != nil { + return nil, fmt.Errorf("decoding n: %w", err) + } + eBytes, err := base64.RawURLEncoding.DecodeString(eStr) + if err != nil { + return nil, fmt.Errorf("decoding e: %w", err) + } + + n := new(big.Int).SetBytes(nBytes) + e := int(new(big.Int).SetBytes(eBytes).Int64()) + + return &rsa.PublicKey{N: n, E: e}, nil +} + + + + + + + +// Authenticate uses the headers from the given request to get the user id. The +// returned context will be cancled, if the session is revoked. +func (a *Auth) AuthenticateLegacy(w http.ResponseWriter, r *http.Request) (context.Context, error) { + if a.fake { + return r.Context(), nil + } + + ctx := r.Context() + p := new(payload) if err := a.loadToken(w, r, p); err != nil { return nil, fmt.Errorf("reading token: %w", err) From 41dd690e523aa67376b4f43a96a4b20292a22bf3 Mon Sep 17 00:00:00 2001 From: Jan Malte Behrens Date: Tue, 19 May 2026 16:40:59 +0200 Subject: [PATCH 02/17] fix: Applying correct rsa parsing method --- auth/auth.go | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/auth/auth.go b/auth/auth.go index ff9ad0f58..b927478ef 100644 --- a/auth/auth.go +++ b/auth/auth.go @@ -151,7 +151,6 @@ func (a *Auth) Authenticate(w http.ResponseWriter, r *http.Request) (context.Con return nil, fmt.Errorf("reading token: %w", err) } - fmt.Println("We are here babeeey") if p.KeycloakID == "" { return a.AuthenticatedContext(ctx, 0), nil } @@ -204,7 +203,12 @@ func (a *Auth) loadTokenKeycloak(w http.ResponseWriter, r *http.Request, payload } _, err = jwt.ParseWithClaims(encodedToken, payload, func(token *jwt.Token) (interface{}, error) { - return []byte(a.tokenKey), nil + kid, ok := token.Header["kid"].(string) + if !ok { + return nil, fmt.Errorf("Missing kid in token header") + } + + return a.getKey(r.Context(), kid) }) if err != nil { From c1772850385f987073f7005e0705164dcf5b5cf8 Mon Sep 17 00:00:00 2001 From: Jan Malte Behrens Date: Tue, 19 May 2026 17:32:15 +0200 Subject: [PATCH 03/17] feat: Removing old auth method code --- auth/auth.go | 207 ++------------------------------------------------- 1 file changed, 8 insertions(+), 199 deletions(-) diff --git a/auth/auth.go b/auth/auth.go index b927478ef..1278c719b 100644 --- a/auth/auth.go +++ b/auth/auth.go @@ -22,20 +22,8 @@ import ( "github.com/ostcar/topic" ) -// DebugTokenKey and DebugCookieKey are non random auth keys for development. -const ( - DebugTokenKey = "auth-dev-token-key" - DebugCookieKey = "auth-dev-cookie-key" -) - var ( - envAuthHost = environment.NewVariable("AUTH_HOST", "localhost", "Host of the auth service.") - envAuthPort = environment.NewVariable("AUTH_PORT", "9004", "Port of the auth service.") - envAuthProtocol = environment.NewVariable("AUTH_PROTOCOL", "http", "Protocol of the auth service.") - envAuthFake = environment.NewVariable("AUTH_FAKE", "false", "Use user id 1 for every request. Ignores all other auth environment variables.") - - envAuthTokenFile = environment.NewVariable("AUTH_TOKEN_KEY_FILE", "/run/secrets/auth_token_key", "Key to sign the JWT auth tocken.") - envAuthCookieFile = environment.NewVariable("AUTH_COOKIE_KEY_FILE", "/run/secrets/auth_cookie_key", "Key to sign the JWT auth cookie.") + envAuthFake = environment.NewVariable("AUTH_FAKE", "false", "Use user id 1 for every request. Ignores all other auth environment variables.") envIssuerURL = environment.NewVariable("OIDC_ISSUER_URL", "http://localhost:8080/realms/openslides", "URL of keycloak server") envIssuerURLDocker = environment.NewVariable("OIDC_ISSUER_URL_DOCKER", "http://keycloak-server:8080/realms/openslides", "Dockerized URL of keycloak server") @@ -45,13 +33,11 @@ var ( ) // pruneTime defines how long a topic id will be valid. This should be higher -// then the max livetime of a token. +// than the max lifetime of a token. const pruneTime = 15 * time.Minute const ( - cookieName = "refreshId" authHeader = "Authorization" - authPath = "/internal/auth/authenticate" ) // LogoutEventer tells, when a sessionID gets revoked. @@ -70,8 +56,6 @@ type Auth struct { logedoutSessions *topic.Topic[string] - authServiceURL string - tokenKey string cookieKey string @@ -86,36 +70,15 @@ type Auth struct { // Returns the initialized Auth objectand a function to be called in the // background. func New(lookup environment.Environmenter, messageBus LogoutEventer) (*Auth, func(context.Context, func(error)), error) { - url := fmt.Sprintf( - "%s://%s:%s", - envAuthProtocol.Value(lookup), - envAuthHost.Value(lookup), - envAuthPort.Value(lookup), - ) - - fake, _ := strconv.ParseBool(envAuthFake.Value(lookup)) - - authToken, err := environment.ReadSecretWithDefault(lookup, envAuthTokenFile, DebugTokenKey) - if err != nil { - return nil, nil, fmt.Errorf("reading auth token: %w", err) - } - - cookieToken, err := environment.ReadSecretWithDefault(lookup, envAuthCookieFile, DebugCookieKey) - if err != nil { - return nil, nil, fmt.Errorf("reading cookie token: %w", err) - } - issuerURLDocker := envIssuerURLDocker.Value(lookup) if issuerURLDocker == "" { issuerURLDocker = envIssuerURL.Value(lookup) } + fake, _ := strconv.ParseBool(envAuthFake.Value(lookup)) + a := &Auth{ - fake: fake, logedoutSessions: topic.New[string](), - authServiceURL: url, - tokenKey: authToken, - cookieKey: cookieToken, issuerURLDocker: issuerURLDocker, keys: make(map[string]*rsa.PublicKey), } @@ -215,7 +178,7 @@ func (a *Auth) loadTokenKeycloak(w http.ResponseWriter, r *http.Request, payload var invalid *jwt.ValidationError if errors.As(err, &invalid) { fmt.Println(err) - return a.handleInvalidToken(r.Context(), invalid, w, encodedToken, "") + return a.handleInvalidToken(r.Context(), invalid, w, encodedToken) } } @@ -359,58 +322,7 @@ func parseRSAPublicKey(nStr, eStr string) (*rsa.PublicKey, error) { return &rsa.PublicKey{N: n, E: e}, nil } - - - - - - -// Authenticate uses the headers from the given request to get the user id. The -// returned context will be cancled, if the session is revoked. -func (a *Auth) AuthenticateLegacy(w http.ResponseWriter, r *http.Request) (context.Context, error) { - if a.fake { - return r.Context(), nil - } - - ctx := r.Context() - - p := new(payload) - if err := a.loadToken(w, r, p); err != nil { - return nil, fmt.Errorf("reading token: %w", err) - } - - if p.UserID == 0 { - return a.AuthenticatedContext(ctx, 0), nil - } - - cid, sessionIDs := a.logedoutSessions.ReceiveAll() - if slices.Contains(sessionIDs, p.SessionID) { - return nil, &authError{"invalid session", nil} - } - - ctx, cancelCtx := context.WithCancel(a.AuthenticatedContext(ctx, p.UserID)) - - go func() { - defer cancelCtx() - - var sessionIDs []string - var err error - for { - cid, sessionIDs, err = a.logedoutSessions.ReceiveSince(ctx, cid) - if err != nil { - return - } - - if slices.Contains(sessionIDs, p.SessionID) { - return - } - } - }() - - return ctx, nil -} - -// AuthenticatedContext returns a new context that contains an userID. +// AuthenticatedContext returns a new context that contains a userID. // // Should only used for internal URLs. All other URLs should use auth.Authenticate. func (a *Auth) AuthenticatedContext(ctx context.Context, userID int) context.Context { @@ -472,124 +384,21 @@ func (a *Auth) pruneOldData(ctx context.Context) { } } -// loadToken loads and validates the token. If the token is expires, it tries -// to renews it and writes the new token to the responsewriter. -func (a *Auth) loadToken(w http.ResponseWriter, r *http.Request, payload jwt.Claims) error { - header := r.Header.Get(authHeader) - cookie, err := r.Cookie(cookieName) - if err != nil && err != http.ErrNoCookie { - return fmt.Errorf("reading cookie: %w", err) - } - - encodedToken := strings.TrimPrefix(header, "bearer ") - - if cookie == nil && header == encodedToken { - // No token and no auth cookie. Handle the request as public access requst. - return nil - } - - if cookie == nil && header != encodedToken { - return authError{"Can not find auth cookie", nil} - } - - if cookie != nil && header == encodedToken { - return authError{"Can not find auth token", nil} - } - - encodedCookie := strings.TrimPrefix(cookie.Value, "bearer%20") - - _, err = jwt.Parse(encodedCookie, func(token *jwt.Token) (interface{}, error) { - return []byte(a.cookieKey), nil - }) - if err != nil { - var invalid *jwt.ValidationError - if errors.As(err, &invalid) { - return authError{"Invalid auth token", err} - } - return fmt.Errorf("validating auth cookie: %w", err) - } - - _, err = jwt.ParseWithClaims(encodedToken, payload, func(token *jwt.Token) (interface{}, error) { - return []byte(a.tokenKey), nil - }) - if err != nil { - var invalid *jwt.ValidationError - if errors.As(err, &invalid) { - return a.handleInvalidToken(r.Context(), invalid, w, encodedToken, encodedCookie) - } - } - - return nil -} - -func (a *Auth) handleInvalidToken(ctx context.Context, invalid *jwt.ValidationError, w http.ResponseWriter, encodedToken, encodedCookie string) error { +func (a *Auth) handleInvalidToken(ctx context.Context, invalid *jwt.ValidationError, w http.ResponseWriter, encodedToken string) error { if !tokenExpired(invalid.Errors) { return authError{"Invalid auth token", invalid} } - token, err := a.refreshToken(ctx, encodedToken, encodedCookie) - if err != nil { - return fmt.Errorf("refreshing token: %w", err) - } - - w.Header().Set(authHeader, token) - return nil + return authError{"Token expired", invalid} } func tokenExpired(errNo uint32) bool { return errNo&(jwt.ValidationErrorExpired|jwt.ValidationErrorNotValidYet) != 0 } -func (a *Auth) refreshToken(ctx context.Context, token, cookie string) (string, error) { - req, err := http.NewRequestWithContext(ctx, "POST", a.authServiceURL+authPath, nil) - if err != nil { - return "", fmt.Errorf("creating auth request: %w", err) - } - - req.Header.Add(authHeader, "bearer "+token) - req.AddCookie(&http.Cookie{Name: cookieName, Value: "bearer " + cookie, HttpOnly: true, Secure: true}) - - resp, err := http.DefaultClient.Do(req) - if err != nil { - // TODO External ERROR - return "", fmt.Errorf("send request to auth service: %w", err) - } - defer resp.Body.Close() - - if resp.StatusCode != 200 { - if resp.StatusCode == 403 { - return "", authError{msg: "Invalid Session", wrapped: err} - } - // TODO LAST ERROR - return "", fmt.Errorf("auth-service returned status %s", resp.Status) - } - - newToken := resp.Header.Get(authHeader) - if newToken == "" { - var rPayload struct { - Message string `json:"message"` - } - if err := json.NewDecoder(resp.Body).Decode(&rPayload); err != nil { - return "", fmt.Errorf("decoding auth response: %w", err) - } - if rPayload.Message == "" { - rPayload.Message = "Can not refresh token" - } - return "", authError{rPayload.Message, nil} - - } - - return newToken, nil -} - type authString string const ( userIDType authString = "user_id" ) -type payload struct { - jwt.StandardClaims - UserID int `json:"userId"` - SessionID string `json:"sessionId"` -} From c91d2f67e4786ce0739d5f7a175ef100bd264409 Mon Sep 17 00:00:00 2001 From: Jan Malte Behrens Date: Fri, 29 May 2026 14:11:21 +0200 Subject: [PATCH 04/17] feat: Use os-id from JWT --- auth/auth.go | 28 +++++++++++++++------------- 1 file changed, 15 insertions(+), 13 deletions(-) diff --git a/auth/auth.go b/auth/auth.go index 1278c719b..8df744211 100644 --- a/auth/auth.go +++ b/auth/auth.go @@ -4,13 +4,13 @@ package auth import ( "context" - "encoding/json" + "crypto/rsa" "encoding/base64" + "encoding/json" "errors" "fmt" "math/big" "net/http" - "crypto/rsa" "slices" "strconv" "strings" @@ -23,13 +23,13 @@ import ( ) var ( - envAuthFake = environment.NewVariable("AUTH_FAKE", "false", "Use user id 1 for every request. Ignores all other auth environment variables.") + envAuthFake = environment.NewVariable("AUTH_FAKE", "false", "Use user id 1 for every request. Ignores all other auth environment variables.") - envIssuerURL = environment.NewVariable("OIDC_ISSUER_URL", "http://localhost:8080/realms/openslides", "URL of keycloak server") - envIssuerURLDocker = environment.NewVariable("OIDC_ISSUER_URL_DOCKER", "http://keycloak-server:8080/realms/openslides", "Dockerized URL of keycloak server") - envClientID = environment.NewVariable("OIDC_CLIENT_ID", "proxy-client", "Keycloak client name") - envClientSecret = environment.NewVariable("OIDC_CLIENT_SECRET", "proxy-secret", "Keycloak client secret") - envSecret = environment.NewVariable("OIDC_SECRET", "qvAcTGWBIGg7aWKCKRyUsTf33jK3lsmK", "Keycloak secret") + envIssuerURL = environment.NewVariable("OIDC_ISSUER_URL", "http://localhost:8080/realms/openslides", "URL of keycloak server") + envIssuerURLDocker = environment.NewVariable("OIDC_ISSUER_URL_DOCKER", "http://keycloak-server:8080/realms/openslides", "Dockerized URL of keycloak server") + envClientID = environment.NewVariable("OIDC_CLIENT_ID", "proxy-client", "Keycloak client name") + envClientSecret = environment.NewVariable("OIDC_CLIENT_SECRET", "proxy-secret", "Keycloak client secret") + envSecret = environment.NewVariable("OIDC_SECRET", "qvAcTGWBIGg7aWKCKRyUsTf33jK3lsmK", "Keycloak secret") ) // pruneTime defines how long a topic id will be valid. This should be higher @@ -80,7 +80,7 @@ func New(lookup environment.Environmenter, messageBus LogoutEventer) (*Auth, fun a := &Auth{ logedoutSessions: topic.New[string](), issuerURLDocker: issuerURLDocker, - keys: make(map[string]*rsa.PublicKey), + keys: make(map[string]*rsa.PublicKey), } // Make sure the topic is not empty @@ -98,7 +98,6 @@ func New(lookup environment.Environmenter, messageBus LogoutEventer) (*Auth, fun return a, background, nil } - // Authenticate uses the headers from the given request to get the user id. The // returned context will be cancled, if the session is revoked. func (a *Auth) Authenticate(w http.ResponseWriter, r *http.Request) (context.Context, error) { @@ -124,7 +123,10 @@ func (a *Auth) Authenticate(w http.ResponseWriter, r *http.Request) (context.Con } // Get OS User Id linked to Keycloak ID - userID := 1 + userID, err := strconv.Atoi(p.OSUserID) + if err != nil { + return nil, &authError{"user id is not an integer " + p.OSUserID, nil} + } ctx, cancelCtx := context.WithCancel(a.AuthenticatedContext(ctx, userID)) @@ -234,6 +236,7 @@ type payloadKeycloak struct { Email string `json:"email"` Username string `json:"preferred_username"` ClientName string `json:"azp"` + OSUserID string `json:"os_id"` } // getKey returns the RSA public key for the given kid, fetching from JWKS if needed @@ -253,7 +256,7 @@ func (a *Auth) fetchJWKS(ctx context.Context, kid string) (*rsa.PublicKey, error return key, nil } - req, err := http.NewRequestWithContext(ctx, "GET", a.issuerURLDocker + "/protocol/openid-connect/certs", nil) + req, err := http.NewRequestWithContext(ctx, "GET", a.issuerURLDocker+"/protocol/openid-connect/certs", nil) if err != nil { return nil, fmt.Errorf("creating JWKS request: %w", err) } @@ -401,4 +404,3 @@ type authString string const ( userIDType authString = "user_id" ) - From 6291545bb440d2f557466d92d0d65333d02819a8 Mon Sep 17 00:00:00 2001 From: Jan Malte Behrens Date: Thu, 11 Jun 2026 15:36:35 +0200 Subject: [PATCH 05/17] feat: Adopt new meta --- meta | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/meta b/meta index 27283ae3a..657f04058 160000 --- a/meta +++ b/meta @@ -1 +1 @@ -Subproject commit 27283ae3afd8e8700be9d979b0ec2d536b04b084 +Subproject commit 657f040586c764da674fce05bff58e13b93ad3a3 From e4986fe17234db72f47782f9f79a10fd87874bfa Mon Sep 17 00:00:00 2001 From: "openslides-automation[bot]" <125256978+openslides-automation[bot]@users.noreply.github.com> Date: Fri, 3 Jul 2026 13:57:43 +0200 Subject: [PATCH 06/17] Update meta repository (#288) --- meta | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/meta b/meta index d3b883cb2..7f0f4fe86 160000 --- a/meta +++ b/meta @@ -1 +1 @@ -Subproject commit d3b883cb2777153dc9a99fc325d62b777be0f220 +Subproject commit 7f0f4fe86d179bc0bf07cecbcce508cec72922f6 From 670d0d5864f101d279cb474439934ed509be8ee2 Mon Sep 17 00:00:00 2001 From: "openslides-automation[bot]" <125256978+openslides-automation[bot]@users.noreply.github.com> Date: Mon, 6 Jul 2026 17:07:09 +0200 Subject: [PATCH 07/17] Update meta repository (#290) Co-authored-by: vkrasnovyd <114735598+vkrasnovyd@users.noreply.github.com> --- datastore/pgtest/sql/base_data.sql | 5 +++++ meta | 2 +- 2 files changed, 6 insertions(+), 1 deletion(-) diff --git a/datastore/pgtest/sql/base_data.sql b/datastore/pgtest/sql/base_data.sql index 1917522f9..5c49f0e86 100644 --- a/datastore/pgtest/sql/base_data.sql +++ b/datastore/pgtest/sql/base_data.sql @@ -165,4 +165,9 @@ projector_t VALUES (1, 1, 'main', 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1); SELECT nextval('projector_t_id_seq'); + +INSERT INTO +version (migration_index, migration_state) +VALUES (100, 'finalized'); + COMMIT; diff --git a/meta b/meta index 7f0f4fe86..8d674fe32 160000 --- a/meta +++ b/meta @@ -1 +1 @@ -Subproject commit 7f0f4fe86d179bc0bf07cecbcce508cec72922f6 +Subproject commit 8d674fe32cbadef2d572f11abf4cfa81f09c94e1 From ae2aa4df5126745b8e9631898473d3dbee1478af Mon Sep 17 00:00:00 2001 From: Bastian Rihm Date: Tue, 7 Jul 2026 15:28:28 +0200 Subject: [PATCH 08/17] Compatibility check workflow (#291) --- .github/workflows/compatibility-check.yml | 77 +++++++++++++++++++++++ 1 file changed, 77 insertions(+) create mode 100644 .github/workflows/compatibility-check.yml diff --git a/.github/workflows/compatibility-check.yml b/.github/workflows/compatibility-check.yml new file mode 100644 index 000000000..9bd016553 --- /dev/null +++ b/.github/workflows/compatibility-check.yml @@ -0,0 +1,77 @@ +name: Compatibility Check + +on: + pull_request: + types: [opened, synchronize, reopened, ready_for_review] + +permissions: + contents: read + +jobs: + compatibility: + runs-on: ubuntu-latest + strategy: + fail-fast: false + matrix: + service: + - openslides-vote-service + - openslides-projector-service + - openslides-search-service + - openslides-autoupdate-service + - openslides-icc-service + + steps: + - name: Checkout openslides-go (PR head) + uses: actions/checkout@v4 + with: + repository: OpenSlides/openslides-go + ref: ${{ github.event.pull_request.head.sha }} + path: openslides-go + submodules: recursive + fetch-depth: 0 + + - name: Determine ref for service repo + id: ref + env: + GH_TOKEN: ${{ secrets.GITHUB_TOKEN }} + OWNER: OpenSlides + REPO: ${{ matrix.service }} + PR_BRANCH: ${{ github.event.pull_request.head.ref }} + run: | + set -euo pipefail + if gh api "repos/${OWNER}/${REPO}/branches/${PR_BRANCH}" >/dev/null 2>&1; then + echo "ref=${PR_BRANCH}" >> "$GITHUB_OUTPUT" + else + default_branch="$(gh api "repos/${OWNER}/${REPO}" --jq .default_branch)" + echo "ref=${default_branch}" >> "$GITHUB_OUTPUT" + fi + + - name: Checkout service repo (recursive submodules) + uses: actions/checkout@v4 + with: + repository: OpenSlides/${{ matrix.service }} + ref: ${{ steps.ref.outputs.ref }} + path: service + submodules: recursive + fetch-depth: 0 + + - name: Set up Go + uses: actions/setup-go@v5 + with: + go-version-file: service/go.mod + cache: true + + - name: Create temporary go.work (service + openslides-go) + run: | + set -euo pipefail + go work init ./service ./openslides-go + # sync is optional but helps keep workspace module graph consistent + go work sync + + - name: Run go vet (service against PR openslides-go) + working-directory: service + run: go vet ./... + + - name: Run go test (service against PR openslides-go) + working-directory: service + run: go test ./... From c412814e5399dd8895901ea6745f042a8ceb2af4 Mon Sep 17 00:00:00 2001 From: Oskar Hahn Date: Wed, 8 Jul 2026 11:24:59 +0200 Subject: [PATCH 09/17] Explain auto docu and fix NewFlowPostgres (#262) --- datastore/flow_postgres.go | 92 +++++++++++++++++++-------------- datastore/flow_postgres_test.go | 20 +++++-- datastore/pgtest/pgtest.go | 20 +++---- environment/README.md | 72 ++++++++++++++++++++++++++ environment/environment.go | 8 ++- 5 files changed, 158 insertions(+), 54 deletions(-) create mode 100644 environment/README.md diff --git a/datastore/flow_postgres.go b/datastore/flow_postgres.go index de994691d..5c31d23cd 100644 --- a/datastore/flow_postgres.go +++ b/datastore/flow_postgres.go @@ -44,33 +44,38 @@ type FlowPostgres struct { } // NewFlowPostgres initializes a SourcePostgres. -func NewFlowPostgres(lookup environment.Environmenter) (*FlowPostgres, error) { +// +// Returns the postgres instance, an init function, that has to be called before +// the first use and an error. +func NewFlowPostgres(lookup environment.Environmenter) (*FlowPostgres, func(context.Context) error, error) { addr, err := postgresDSN(lookup) if err != nil { - return nil, fmt.Errorf("reading postgres dns: %w", err) + return nil, nil, fmt.Errorf("reading postgres dns: %w", err) } config, err := pgxpool.ParseConfig(addr) if err != nil { - return nil, fmt.Errorf("parse config: %w", err) + return nil, nil, fmt.Errorf("parse config: %w", err) } config.ConnConfig.DefaultQueryExecMode = pgx.QueryExecModeSimpleProtocol - ctx := context.Background() - pool, err := pgxpool.NewWithConfig(ctx, config) + // NewWithConfig does no IO, if MinConns == 0. So the background-context + // does nothing in this case. + config.MinConns = 0 + pool, err := pgxpool.NewWithConfig(context.Background(), config) if err != nil { - return nil, fmt.Errorf("creating connection pool: %w", err) + return nil, nil, fmt.Errorf("creating connection pool: %w", err) } notifyAddr, err := postgresDSNNotify(lookup) if err != nil { - return nil, fmt.Errorf("reading postgres dns for notify: %w", err) + return nil, nil, fmt.Errorf("reading postgres dns for notify: %w", err) } notifyConf, err := pgx.ParseConfig(notifyAddr) if err != nil { - return nil, fmt.Errorf("generate config for notify: %w", err) + return nil, nil, fmt.Errorf("generate config for notify: %w", err) } flow := FlowPostgres{ @@ -78,24 +83,32 @@ func NewFlowPostgres(lookup environment.Environmenter) (*FlowPostgres, error) { notifyConfig: notifyConf, } - if err := flow.updateEnums(ctx); err != nil { - return nil, err + init := func(ctx context.Context) error { + if err := waitPostgresAvailable(ctx, config.ConnConfig); err != nil { + return fmt.Errorf("waiting for postgres: %w", err) + } + + if err := flow.updateEnums(ctx); err != nil { + return fmt.Errorf("init enum values: %w", err) + } + return nil } - return &flow, nil + return &flow, init, nil } +// updateEnums ready the enum values from postgres for later use. func (p *FlowPostgres) updateEnums(ctx context.Context) error { c, err := p.Pool.Acquire(ctx) if err != nil { - return err + return fmt.Errorf("acquire connection: %w", err) } defer c.Release() sql := `SELECT oid, typarray FROM pg_type WHERE typtype = 'e';` rows, err := c.Conn().Query(ctx, sql) if err != nil { - return err + return fmt.Errorf("fetch pg_types: %w", err) } defer rows.Close() @@ -105,7 +118,7 @@ func (p *FlowPostgres) updateEnums(ctx context.Context) error { var oid uint32 var typarray uint32 if err := rows.Scan(&oid, &typarray); err != nil { - return err + return fmt.Errorf("read type: %w", err) } p.enums[oid] = struct{}{} @@ -314,13 +327,17 @@ func (p *FlowPostgres) Update(ctx context.Context, updateFn func(map[dskey.Key][ } if conn == nil || conn.IsClosed() { - conn = getPostgresConnection(ctx, p.notifyConfig) + var err error + conn, err = getPostgresConnection(ctx, p.notifyConfig) + if err != nil { + updateFn(nil, fmt.Errorf("get postgres connection: %w", err)) + continue + } if lastXactID > 0 { oslog.Info("Database reconnected") } - _, err := conn.Exec(ctx, "LISTEN os_notify") - if err != nil { + if _, err := conn.Exec(ctx, "LISTEN os_notify"); err != nil { updateFn(nil, fmt.Errorf("listen on channel os_notify: %w", err)) continue } @@ -421,27 +438,13 @@ func (p *FlowPostgres) Update(ctx context.Context, updateFn func(map[dskey.Key][ } } -// WaitPostgresAvailable blocks until postgres db is availabe -func WaitPostgresAvailable(lookup environment.Environmenter) error { - if _, forDocu := lookup.(*environment.ForDocu); forDocu { - return nil - } - - addr, err := postgresDSN(lookup) - if err != nil { - return fmt.Errorf("reading postgres config: %w", err) - } - - config, err := pgx.ParseConfig(addr) - if err != nil { - return fmt.Errorf("parsing postgres config: %w", err) - } - - ctx := context.Background() - +func waitPostgresAvailable(ctx context.Context, config *pgx.ConnConfig) error { for { - conn := getPostgresConnection(ctx, config) - err := waitDatabaseInitialized(ctx, conn) + conn, err := getPostgresConnection(ctx, config) + if err != nil { + return fmt.Errorf("get postgres connection: %w", err) + } + err = waitDatabaseInitialized(ctx, conn) _ = conn.Close(ctx) if err != nil { time.Sleep(1 * time.Second) @@ -453,13 +456,21 @@ func WaitPostgresAvailable(lookup environment.Environmenter) error { } // getPostgresConnection tries to connect to a database until it is successful -// TODO: Return error on credential related errors -func getPostgresConnection(ctx context.Context, connConfig *pgx.ConnConfig) *pgx.Conn { +// TODO: Only returny on some known errors. +func getPostgresConnection(ctx context.Context, connConfig *pgx.ConnConfig) (*pgx.Conn, error) { retryDelay := 1 * time.Second for { conn, err := pgx.ConnectConfig(ctx, connConfig) if err != nil { + if errors.Is(err, context.DeadlineExceeded) { + // TODO: This is inverted. The function should only retry on + // some known error and return on any unknown error. But I don't + // know what the error is, on which the service should retry. If + // nobody knows, we should fail on any error and wait for + // production to report the correct error. + return nil, fmt.Errorf("pgx connect: %w", err) + } oslog.Error("Error connecting to db: %v", err) time.Sleep(retryDelay) continue @@ -468,13 +479,14 @@ func getPostgresConnection(ctx context.Context, connConfig *pgx.ConnConfig) *pgx pingCtx, cancel := context.WithTimeout(ctx, 5*time.Second) err = conn.Ping(pingCtx) if err != nil { + cancel() oslog.Info("Waiting for db to become ready") time.Sleep(retryDelay) continue } cancel() - return conn + return conn, nil } } diff --git a/datastore/flow_postgres_test.go b/datastore/flow_postgres_test.go index e3616734f..37eea5654 100644 --- a/datastore/flow_postgres_test.go +++ b/datastore/flow_postgres_test.go @@ -133,11 +133,14 @@ func TestFlowPostgres(t *testing.T) { t.Fatalf("adding example data: %v", pgtest.PrityPostgresError(err, tt.insert)) } - flow, err := datastore.NewFlowPostgres(environment.ForTests(tp.Env)) + flow, init, err := datastore.NewFlowPostgres(environment.ForTests(tp.Env)) if err != nil { t.Fatalf("NewFlowPostgres(): %v", err) } defer flow.Close() + if err := init(ctx); err != nil { + t.Fatalf("init postgres: %v", err) + } keys := make([]dskey.Key, 0, len(tt.expect)) for k := range tt.expect { @@ -180,10 +183,13 @@ func TestPostgresUpdate(t *testing.T) { t.Fatalf("create connection: %v", err) } - flow, err := datastore.NewFlowPostgres(environment.ForTests(tp.Env)) + flow, init, err := datastore.NewFlowPostgres(environment.ForTests(tp.Env)) if err != nil { t.Fatalf("NewFlowPostgres(): %v", err) } + if err := init(ctx); err != nil { + t.Fatalf("init postgres: %v", err) + } keys := []dskey.Key{ dskey.MustKey("user/300/username"), @@ -253,10 +259,13 @@ func TestPostgresUpdateCollectionWithCalculatedField(t *testing.T) { t.Fatalf("create connection: %v", err) } - flow, err := datastore.NewFlowPostgres(environment.ForTests(tp.Env)) + flow, init, err := datastore.NewFlowPostgres(environment.ForTests(tp.Env)) if err != nil { t.Fatalf("NewFlowPostgres(): %v", err) } + if err := init(ctx); err != nil { + t.Fatalf("init postgres: %v", err) + } keys := []dskey.Key{ dskey.MustKey("poll/300/title"), @@ -358,10 +367,13 @@ func TestBigQuery(t *testing.T) { t.Fatalf("starting postgres: %v", err) } - flow, err := datastore.NewFlowPostgres(environment.ForTests(tp.Env)) + flow, init, err := datastore.NewFlowPostgres(environment.ForTests(tp.Env)) if err != nil { t.Fatalf("NewSource(): %v", err) } + if err := init(ctx); err != nil { + t.Fatalf("init postgres: %v", err) + } conn, err := tp.Conn(ctx) if err != nil { diff --git a/datastore/pgtest/pgtest.go b/datastore/pgtest/pgtest.go index 1b49028b8..9955e0ef6 100644 --- a/datastore/pgtest/pgtest.go +++ b/datastore/pgtest/pgtest.go @@ -21,7 +21,7 @@ import ( // But for security reasons, it can not include files outside the directory. // Therefore it is necessary to copry the files to create the sql-schema to this // folder. The embedding is necessary, to other repositories like the -// vote-service, do not need to include the meta-repo as a sub repo. +// vote-service, to not need to include the meta-repo as a sub repo. //go:generate go run ./copy_sql @@ -41,13 +41,12 @@ func RunTests(m *testing.M) int { pool, err = dockertest.NewPool(ctx, "", dockertest.WithMaxWait(2*time.Minute), ) + defer pool.Close(ctx) if err != nil { - panic(err) + fmt.Printf("Error: %v", err) + return 1 } code := m.Run() - // Close removes all tracked containers/networks and closes the client. - // Call before os.Exit — deferred functions do not run after os.Exit. - pool.Close(ctx) return code } @@ -112,7 +111,7 @@ func NewPostgresTest(t *testing.T) (*PostgresTest, error) { return nil, fmt.Errorf("create test database: %w", err) } - addr := fmt.Sprintf(`user=postgres password='openslides' host=localhost port=%s dbname=%s`, port, database) + addr := fmt.Sprintf(`user=postgres password='openslides' host=127.0.0.1 port=%s dbname=%s`, port, database) config, err := pgx.ParseConfig(addr) if err != nil { return nil, fmt.Errorf("parse config: %w", err) @@ -122,7 +121,7 @@ func NewPostgresTest(t *testing.T) (*PostgresTest, error) { pgxConfig: config, Env: map[string]string{ - "DATABASE_HOST": "localhost", + "DATABASE_HOST": "127.0.0.1", "DATABASE_PORT": port, "DATABASE_NAME": database, "DATABASE_USER": "postgres", @@ -222,11 +221,14 @@ func (tp *PostgresTest) AddData(ctx context.Context, data string) error { } // Flow returns a flow that is using the postgres instance. -func (tp *PostgresTest) Flow() (*datastore.FlowPostgres, error) { - flow, err := datastore.NewFlowPostgres(environment.ForTests(tp.Env)) +func (tp *PostgresTest) Flow(ctx context.Context) (*datastore.FlowPostgres, error) { + flow, init, err := datastore.NewFlowPostgres(environment.ForTests(tp.Env)) if err != nil { return nil, fmt.Errorf("create postgres flow: %w", err) } + if err := init(ctx); err != nil { + return nil, fmt.Errorf("init postgres: %w", err) + } return flow, nil } diff --git a/environment/README.md b/environment/README.md new file mode 100644 index 000000000..43b743e38 --- /dev/null +++ b/environment/README.md @@ -0,0 +1,72 @@ +# Environment + +The environment package is used, to configure a go-service. The Package contains +functions, to access environment variables in a way, so that the configuration +for a service can automaticly be generated. + +For this to work, the initalization phase is not allowed to do any IO. In other +case, the IO would also happen, then the docutmentation is build. + +Therefore, a service service has to start in different phases. + +The first phase ready the environment variables, but does nothing else. For +example, it could be build like this: + +```go +func initService(lookup environment.Environmenter) (func(context.Context) error, error) { + // No IO allowed in this function + oslog.InitLog(lookup) + someType, background := somePackage.New(lookup) + + service := func(ctx context.Context) error { + // This is a callback. IO is allowed here. + + oslog.Info("Listen on %s", listenAddr) + return http.Run(ctx, someType) + } + return service, nil +} +``` + +All functions inside initService (like `somePackage.New`) are also not allowed +to do IO. If IO is needed, it has to be done in the callback. + + +When this is done, the service can either be started: + +```go +func run(ctx context.Context) error { + lookup := new(environment.ForProduction) + + service, err := initService(lookup) + if err != nil { + return fmt.Errorf("init services: %w", err) + } + + return service(ctx) +} +``` + +Or the documentation can be build: + +```go +func buildDocu() error { + lookup := new(environment.ForDocu) + + if _, err := initService(lookup); err != nil { + return fmt.Errorf("init services: %w", err) + } + + doc, err := lookup.BuildDoc() + if err != nil { + return fmt.Errorf("build doc: %w", err) + } + + fmt.Println(doc) + return nil +} +``` + +As a maxim: If the function has a `context.Context` as an argument, you can do +IO. If the function has not a `context.Context` as an argument, do not create +one with `context.Background()`. You have to do the IO elsewhere. diff --git a/environment/environment.go b/environment/environment.go index 33272ecc0..8c54f70a1 100644 --- a/environment/environment.go +++ b/environment/environment.go @@ -68,7 +68,13 @@ func (v Variable) ValueOr(loopup Environmenter, other Variable) string { // Environmenter is an type, that can return the value for environment // variables. // -// It also saves the used environment variables. +// This type is used to access environment variables, but also, to autogenerate, +// which environment variables a service needs. +// +// For that to work, a function, that takes this type is only allowed to +// initialize datastructures. It is not allowed to do any IO or other blocking +// function calls. Especially, not network request like waiting for postgres. If +// IO is needed, it has to be done in a callback. type Environmenter interface { Getenv(key string) string UseVariable(v Variable) From 945f2e77b1ec49996f64eff555fb661a229271d1 Mon Sep 17 00:00:00 2001 From: Bastian Rihm Date: Wed, 8 Jul 2026 12:09:42 +0200 Subject: [PATCH 10/17] Add openslides-cli to compat check workflow (#293) --- .github/workflows/compatibility-check.yml | 1 + 1 file changed, 1 insertion(+) diff --git a/.github/workflows/compatibility-check.yml b/.github/workflows/compatibility-check.yml index 9bd016553..095757ad0 100644 --- a/.github/workflows/compatibility-check.yml +++ b/.github/workflows/compatibility-check.yml @@ -19,6 +19,7 @@ jobs: - openslides-search-service - openslides-autoupdate-service - openslides-icc-service + - openslides-cli steps: - name: Checkout openslides-go (PR head) From caf2db4304257974819e7a87ec6372ecc9179c63 Mon Sep 17 00:00:00 2001 From: Bastian Rihm Date: Wed, 8 Jul 2026 14:29:30 +0200 Subject: [PATCH 11/17] Revert "Explain auto docu and fix NewFlowPostgres (#262)" (#294) This reverts commit c412814e5399dd8895901ea6745f042a8ceb2af4. --- datastore/flow_postgres.go | 92 ++++++++++++++------------------- datastore/flow_postgres_test.go | 20 ++----- datastore/pgtest/pgtest.go | 20 ++++--- environment/README.md | 72 -------------------------- environment/environment.go | 8 +-- 5 files changed, 54 insertions(+), 158 deletions(-) delete mode 100644 environment/README.md diff --git a/datastore/flow_postgres.go b/datastore/flow_postgres.go index 5c31d23cd..de994691d 100644 --- a/datastore/flow_postgres.go +++ b/datastore/flow_postgres.go @@ -44,38 +44,33 @@ type FlowPostgres struct { } // NewFlowPostgres initializes a SourcePostgres. -// -// Returns the postgres instance, an init function, that has to be called before -// the first use and an error. -func NewFlowPostgres(lookup environment.Environmenter) (*FlowPostgres, func(context.Context) error, error) { +func NewFlowPostgres(lookup environment.Environmenter) (*FlowPostgres, error) { addr, err := postgresDSN(lookup) if err != nil { - return nil, nil, fmt.Errorf("reading postgres dns: %w", err) + return nil, fmt.Errorf("reading postgres dns: %w", err) } config, err := pgxpool.ParseConfig(addr) if err != nil { - return nil, nil, fmt.Errorf("parse config: %w", err) + return nil, fmt.Errorf("parse config: %w", err) } config.ConnConfig.DefaultQueryExecMode = pgx.QueryExecModeSimpleProtocol - // NewWithConfig does no IO, if MinConns == 0. So the background-context - // does nothing in this case. - config.MinConns = 0 - pool, err := pgxpool.NewWithConfig(context.Background(), config) + ctx := context.Background() + pool, err := pgxpool.NewWithConfig(ctx, config) if err != nil { - return nil, nil, fmt.Errorf("creating connection pool: %w", err) + return nil, fmt.Errorf("creating connection pool: %w", err) } notifyAddr, err := postgresDSNNotify(lookup) if err != nil { - return nil, nil, fmt.Errorf("reading postgres dns for notify: %w", err) + return nil, fmt.Errorf("reading postgres dns for notify: %w", err) } notifyConf, err := pgx.ParseConfig(notifyAddr) if err != nil { - return nil, nil, fmt.Errorf("generate config for notify: %w", err) + return nil, fmt.Errorf("generate config for notify: %w", err) } flow := FlowPostgres{ @@ -83,32 +78,24 @@ func NewFlowPostgres(lookup environment.Environmenter) (*FlowPostgres, func(cont notifyConfig: notifyConf, } - init := func(ctx context.Context) error { - if err := waitPostgresAvailable(ctx, config.ConnConfig); err != nil { - return fmt.Errorf("waiting for postgres: %w", err) - } - - if err := flow.updateEnums(ctx); err != nil { - return fmt.Errorf("init enum values: %w", err) - } - return nil + if err := flow.updateEnums(ctx); err != nil { + return nil, err } - return &flow, init, nil + return &flow, nil } -// updateEnums ready the enum values from postgres for later use. func (p *FlowPostgres) updateEnums(ctx context.Context) error { c, err := p.Pool.Acquire(ctx) if err != nil { - return fmt.Errorf("acquire connection: %w", err) + return err } defer c.Release() sql := `SELECT oid, typarray FROM pg_type WHERE typtype = 'e';` rows, err := c.Conn().Query(ctx, sql) if err != nil { - return fmt.Errorf("fetch pg_types: %w", err) + return err } defer rows.Close() @@ -118,7 +105,7 @@ func (p *FlowPostgres) updateEnums(ctx context.Context) error { var oid uint32 var typarray uint32 if err := rows.Scan(&oid, &typarray); err != nil { - return fmt.Errorf("read type: %w", err) + return err } p.enums[oid] = struct{}{} @@ -327,17 +314,13 @@ func (p *FlowPostgres) Update(ctx context.Context, updateFn func(map[dskey.Key][ } if conn == nil || conn.IsClosed() { - var err error - conn, err = getPostgresConnection(ctx, p.notifyConfig) - if err != nil { - updateFn(nil, fmt.Errorf("get postgres connection: %w", err)) - continue - } + conn = getPostgresConnection(ctx, p.notifyConfig) if lastXactID > 0 { oslog.Info("Database reconnected") } - if _, err := conn.Exec(ctx, "LISTEN os_notify"); err != nil { + _, err := conn.Exec(ctx, "LISTEN os_notify") + if err != nil { updateFn(nil, fmt.Errorf("listen on channel os_notify: %w", err)) continue } @@ -438,13 +421,27 @@ func (p *FlowPostgres) Update(ctx context.Context, updateFn func(map[dskey.Key][ } } -func waitPostgresAvailable(ctx context.Context, config *pgx.ConnConfig) error { +// WaitPostgresAvailable blocks until postgres db is availabe +func WaitPostgresAvailable(lookup environment.Environmenter) error { + if _, forDocu := lookup.(*environment.ForDocu); forDocu { + return nil + } + + addr, err := postgresDSN(lookup) + if err != nil { + return fmt.Errorf("reading postgres config: %w", err) + } + + config, err := pgx.ParseConfig(addr) + if err != nil { + return fmt.Errorf("parsing postgres config: %w", err) + } + + ctx := context.Background() + for { - conn, err := getPostgresConnection(ctx, config) - if err != nil { - return fmt.Errorf("get postgres connection: %w", err) - } - err = waitDatabaseInitialized(ctx, conn) + conn := getPostgresConnection(ctx, config) + err := waitDatabaseInitialized(ctx, conn) _ = conn.Close(ctx) if err != nil { time.Sleep(1 * time.Second) @@ -456,21 +453,13 @@ func waitPostgresAvailable(ctx context.Context, config *pgx.ConnConfig) error { } // getPostgresConnection tries to connect to a database until it is successful -// TODO: Only returny on some known errors. -func getPostgresConnection(ctx context.Context, connConfig *pgx.ConnConfig) (*pgx.Conn, error) { +// TODO: Return error on credential related errors +func getPostgresConnection(ctx context.Context, connConfig *pgx.ConnConfig) *pgx.Conn { retryDelay := 1 * time.Second for { conn, err := pgx.ConnectConfig(ctx, connConfig) if err != nil { - if errors.Is(err, context.DeadlineExceeded) { - // TODO: This is inverted. The function should only retry on - // some known error and return on any unknown error. But I don't - // know what the error is, on which the service should retry. If - // nobody knows, we should fail on any error and wait for - // production to report the correct error. - return nil, fmt.Errorf("pgx connect: %w", err) - } oslog.Error("Error connecting to db: %v", err) time.Sleep(retryDelay) continue @@ -479,14 +468,13 @@ func getPostgresConnection(ctx context.Context, connConfig *pgx.ConnConfig) (*pg pingCtx, cancel := context.WithTimeout(ctx, 5*time.Second) err = conn.Ping(pingCtx) if err != nil { - cancel() oslog.Info("Waiting for db to become ready") time.Sleep(retryDelay) continue } cancel() - return conn, nil + return conn } } diff --git a/datastore/flow_postgres_test.go b/datastore/flow_postgres_test.go index 37eea5654..e3616734f 100644 --- a/datastore/flow_postgres_test.go +++ b/datastore/flow_postgres_test.go @@ -133,14 +133,11 @@ func TestFlowPostgres(t *testing.T) { t.Fatalf("adding example data: %v", pgtest.PrityPostgresError(err, tt.insert)) } - flow, init, err := datastore.NewFlowPostgres(environment.ForTests(tp.Env)) + flow, err := datastore.NewFlowPostgres(environment.ForTests(tp.Env)) if err != nil { t.Fatalf("NewFlowPostgres(): %v", err) } defer flow.Close() - if err := init(ctx); err != nil { - t.Fatalf("init postgres: %v", err) - } keys := make([]dskey.Key, 0, len(tt.expect)) for k := range tt.expect { @@ -183,13 +180,10 @@ func TestPostgresUpdate(t *testing.T) { t.Fatalf("create connection: %v", err) } - flow, init, err := datastore.NewFlowPostgres(environment.ForTests(tp.Env)) + flow, err := datastore.NewFlowPostgres(environment.ForTests(tp.Env)) if err != nil { t.Fatalf("NewFlowPostgres(): %v", err) } - if err := init(ctx); err != nil { - t.Fatalf("init postgres: %v", err) - } keys := []dskey.Key{ dskey.MustKey("user/300/username"), @@ -259,13 +253,10 @@ func TestPostgresUpdateCollectionWithCalculatedField(t *testing.T) { t.Fatalf("create connection: %v", err) } - flow, init, err := datastore.NewFlowPostgres(environment.ForTests(tp.Env)) + flow, err := datastore.NewFlowPostgres(environment.ForTests(tp.Env)) if err != nil { t.Fatalf("NewFlowPostgres(): %v", err) } - if err := init(ctx); err != nil { - t.Fatalf("init postgres: %v", err) - } keys := []dskey.Key{ dskey.MustKey("poll/300/title"), @@ -367,13 +358,10 @@ func TestBigQuery(t *testing.T) { t.Fatalf("starting postgres: %v", err) } - flow, init, err := datastore.NewFlowPostgres(environment.ForTests(tp.Env)) + flow, err := datastore.NewFlowPostgres(environment.ForTests(tp.Env)) if err != nil { t.Fatalf("NewSource(): %v", err) } - if err := init(ctx); err != nil { - t.Fatalf("init postgres: %v", err) - } conn, err := tp.Conn(ctx) if err != nil { diff --git a/datastore/pgtest/pgtest.go b/datastore/pgtest/pgtest.go index 9955e0ef6..1b49028b8 100644 --- a/datastore/pgtest/pgtest.go +++ b/datastore/pgtest/pgtest.go @@ -21,7 +21,7 @@ import ( // But for security reasons, it can not include files outside the directory. // Therefore it is necessary to copry the files to create the sql-schema to this // folder. The embedding is necessary, to other repositories like the -// vote-service, to not need to include the meta-repo as a sub repo. +// vote-service, do not need to include the meta-repo as a sub repo. //go:generate go run ./copy_sql @@ -41,12 +41,13 @@ func RunTests(m *testing.M) int { pool, err = dockertest.NewPool(ctx, "", dockertest.WithMaxWait(2*time.Minute), ) - defer pool.Close(ctx) if err != nil { - fmt.Printf("Error: %v", err) - return 1 + panic(err) } code := m.Run() + // Close removes all tracked containers/networks and closes the client. + // Call before os.Exit — deferred functions do not run after os.Exit. + pool.Close(ctx) return code } @@ -111,7 +112,7 @@ func NewPostgresTest(t *testing.T) (*PostgresTest, error) { return nil, fmt.Errorf("create test database: %w", err) } - addr := fmt.Sprintf(`user=postgres password='openslides' host=127.0.0.1 port=%s dbname=%s`, port, database) + addr := fmt.Sprintf(`user=postgres password='openslides' host=localhost port=%s dbname=%s`, port, database) config, err := pgx.ParseConfig(addr) if err != nil { return nil, fmt.Errorf("parse config: %w", err) @@ -121,7 +122,7 @@ func NewPostgresTest(t *testing.T) (*PostgresTest, error) { pgxConfig: config, Env: map[string]string{ - "DATABASE_HOST": "127.0.0.1", + "DATABASE_HOST": "localhost", "DATABASE_PORT": port, "DATABASE_NAME": database, "DATABASE_USER": "postgres", @@ -221,14 +222,11 @@ func (tp *PostgresTest) AddData(ctx context.Context, data string) error { } // Flow returns a flow that is using the postgres instance. -func (tp *PostgresTest) Flow(ctx context.Context) (*datastore.FlowPostgres, error) { - flow, init, err := datastore.NewFlowPostgres(environment.ForTests(tp.Env)) +func (tp *PostgresTest) Flow() (*datastore.FlowPostgres, error) { + flow, err := datastore.NewFlowPostgres(environment.ForTests(tp.Env)) if err != nil { return nil, fmt.Errorf("create postgres flow: %w", err) } - if err := init(ctx); err != nil { - return nil, fmt.Errorf("init postgres: %w", err) - } return flow, nil } diff --git a/environment/README.md b/environment/README.md deleted file mode 100644 index 43b743e38..000000000 --- a/environment/README.md +++ /dev/null @@ -1,72 +0,0 @@ -# Environment - -The environment package is used, to configure a go-service. The Package contains -functions, to access environment variables in a way, so that the configuration -for a service can automaticly be generated. - -For this to work, the initalization phase is not allowed to do any IO. In other -case, the IO would also happen, then the docutmentation is build. - -Therefore, a service service has to start in different phases. - -The first phase ready the environment variables, but does nothing else. For -example, it could be build like this: - -```go -func initService(lookup environment.Environmenter) (func(context.Context) error, error) { - // No IO allowed in this function - oslog.InitLog(lookup) - someType, background := somePackage.New(lookup) - - service := func(ctx context.Context) error { - // This is a callback. IO is allowed here. - - oslog.Info("Listen on %s", listenAddr) - return http.Run(ctx, someType) - } - return service, nil -} -``` - -All functions inside initService (like `somePackage.New`) are also not allowed -to do IO. If IO is needed, it has to be done in the callback. - - -When this is done, the service can either be started: - -```go -func run(ctx context.Context) error { - lookup := new(environment.ForProduction) - - service, err := initService(lookup) - if err != nil { - return fmt.Errorf("init services: %w", err) - } - - return service(ctx) -} -``` - -Or the documentation can be build: - -```go -func buildDocu() error { - lookup := new(environment.ForDocu) - - if _, err := initService(lookup); err != nil { - return fmt.Errorf("init services: %w", err) - } - - doc, err := lookup.BuildDoc() - if err != nil { - return fmt.Errorf("build doc: %w", err) - } - - fmt.Println(doc) - return nil -} -``` - -As a maxim: If the function has a `context.Context` as an argument, you can do -IO. If the function has not a `context.Context` as an argument, do not create -one with `context.Background()`. You have to do the IO elsewhere. diff --git a/environment/environment.go b/environment/environment.go index 8c54f70a1..33272ecc0 100644 --- a/environment/environment.go +++ b/environment/environment.go @@ -68,13 +68,7 @@ func (v Variable) ValueOr(loopup Environmenter, other Variable) string { // Environmenter is an type, that can return the value for environment // variables. // -// This type is used to access environment variables, but also, to autogenerate, -// which environment variables a service needs. -// -// For that to work, a function, that takes this type is only allowed to -// initialize datastructures. It is not allowed to do any IO or other blocking -// function calls. Especially, not network request like waiting for postgres. If -// IO is needed, it has to be done in a callback. +// It also saves the used environment variables. type Environmenter interface { Getenv(key string) string UseVariable(v Variable) From e4132fa01d8cf45fb54d208dbf8424d189a838e0 Mon Sep 17 00:00:00 2001 From: Jan Malte Behrens Date: Mon, 13 Jul 2026 10:12:37 +0200 Subject: [PATCH 12/17] fix: Keycloak to Zitadel changes --- auth/auth.go | 11 ++++------- 1 file changed, 4 insertions(+), 7 deletions(-) diff --git a/auth/auth.go b/auth/auth.go index 8df744211..09f3136db 100644 --- a/auth/auth.go +++ b/auth/auth.go @@ -25,11 +25,8 @@ import ( var ( envAuthFake = environment.NewVariable("AUTH_FAKE", "false", "Use user id 1 for every request. Ignores all other auth environment variables.") - envIssuerURL = environment.NewVariable("OIDC_ISSUER_URL", "http://localhost:8080/realms/openslides", "URL of keycloak server") - envIssuerURLDocker = environment.NewVariable("OIDC_ISSUER_URL_DOCKER", "http://keycloak-server:8080/realms/openslides", "Dockerized URL of keycloak server") - envClientID = environment.NewVariable("OIDC_CLIENT_ID", "proxy-client", "Keycloak client name") - envClientSecret = environment.NewVariable("OIDC_CLIENT_SECRET", "proxy-secret", "Keycloak client secret") - envSecret = environment.NewVariable("OIDC_SECRET", "qvAcTGWBIGg7aWKCKRyUsTf33jK3lsmK", "Keycloak secret") + envIssuerURL = environment.NewVariable("IDP_URL_EXTERNAL", "http://localhost:8080/realms/openslides", "URL of idp server") + envIssuerURLInternal = environment.NewVariable("IDP_URL_INTERNAL", "http://zitadel-api:8080/realms/openslides", "Internal URL of idp server") ) // pruneTime defines how long a topic id will be valid. This should be higher @@ -70,7 +67,7 @@ type Auth struct { // Returns the initialized Auth objectand a function to be called in the // background. func New(lookup environment.Environmenter, messageBus LogoutEventer) (*Auth, func(context.Context, func(error)), error) { - issuerURLDocker := envIssuerURLDocker.Value(lookup) + issuerURLDocker := envIssuerURLInternal.Value(lookup) if issuerURLDocker == "" { issuerURLDocker = envIssuerURL.Value(lookup) } @@ -256,7 +253,7 @@ func (a *Auth) fetchJWKS(ctx context.Context, kid string) (*rsa.PublicKey, error return key, nil } - req, err := http.NewRequestWithContext(ctx, "GET", a.issuerURLDocker+"/protocol/openid-connect/certs", nil) + req, err := http.NewRequestWithContext(ctx, "GET", a.issuerURLDocker+"/oauth/v2/keys", nil) if err != nil { return nil, fmt.Errorf("creating JWKS request: %w", err) } From 1cfc7412a49cac805e99bb645e880b012c5126f9 Mon Sep 17 00:00:00 2001 From: Jan Malte Behrens Date: Mon, 13 Jul 2026 10:38:04 +0200 Subject: [PATCH 13/17] feat: Further renaming of keycloak to zitadel. Readding issuer url check --- auth/auth.go | 32 ++++++++++++++++---------------- 1 file changed, 16 insertions(+), 16 deletions(-) diff --git a/auth/auth.go b/auth/auth.go index 09f3136db..90dcf9f7e 100644 --- a/auth/auth.go +++ b/auth/auth.go @@ -25,8 +25,8 @@ import ( var ( envAuthFake = environment.NewVariable("AUTH_FAKE", "false", "Use user id 1 for every request. Ignores all other auth environment variables.") - envIssuerURL = environment.NewVariable("IDP_URL_EXTERNAL", "http://localhost:8080/realms/openslides", "URL of idp server") - envIssuerURLInternal = environment.NewVariable("IDP_URL_INTERNAL", "http://zitadel-api:8080/realms/openslides", "Internal URL of idp server") + envIssuerURL = environment.NewVariable("IDP_URL_EXTERNAL", "http://localhost:8080", "URL of idp server") + envIssuerURLInternal = environment.NewVariable("IDP_URL_INTERNAL", "http://zitadel-api:8080", "Internal URL of idp server") ) // pruneTime defines how long a topic id will be valid. This should be higher @@ -104,13 +104,13 @@ func (a *Auth) Authenticate(w http.ResponseWriter, r *http.Request) (context.Con ctx := r.Context() - p := new(payloadKeycloak) - if err := a.loadTokenKeycloak(w, r, p); err != nil { + p := new(payloadIDP) + if err := a.loadTokenIDP(w, r, p); err != nil { fmt.Println("reading token: %w", err) return nil, fmt.Errorf("reading token: %w", err) } - if p.KeycloakID == "" { + if p.IDPID == "" { return a.AuthenticatedContext(ctx, 0), nil } @@ -119,7 +119,7 @@ func (a *Auth) Authenticate(w http.ResponseWriter, r *http.Request) (context.Con return nil, &authError{"invalid session", nil} } - // Get OS User Id linked to Keycloak ID + // Get OS User Id linked to IDP ID userID, err := strconv.Atoi(p.OSUserID) if err != nil { return nil, &authError{"user id is not an integer " + p.OSUserID, nil} @@ -149,7 +149,7 @@ func (a *Auth) Authenticate(w http.ResponseWriter, r *http.Request) (context.Con // loadToken loads and validates the token. If the token is expired, it tries // to renew it and write the new token in the responsewriter. -func (a *Auth) loadTokenKeycloak(w http.ResponseWriter, r *http.Request, payload jwt.Claims) error { +func (a *Auth) loadTokenIDP(w http.ResponseWriter, r *http.Request, payload jwt.Claims) error { header := r.Header.Get(authHeader) encodedToken := strings.TrimPrefix(header, "Bearer: ") @@ -184,9 +184,9 @@ func (a *Auth) loadTokenKeycloak(w http.ResponseWriter, r *http.Request, payload return nil } -func (a *Auth) validateToken(ctx context.Context, tokenString string) (*payloadKeycloak, error) { +func (a *Auth) validateToken(ctx context.Context, tokenString string) (*payloadIDP, error) { // 1. Parse token without validation to get kid - token, _, err := new(jwt.Parser).ParseUnverified(tokenString, &payloadKeycloak{}) + token, _, err := new(jwt.Parser).ParseUnverified(tokenString, &payloadIDP{}) if err != nil { return nil, fmt.Errorf("parsing token: %w", err) } @@ -203,7 +203,7 @@ func (a *Auth) validateToken(ctx context.Context, tokenString string) (*payloadK } // 3. Validate token with public key - claims := &payloadKeycloak{} + claims := &payloadIDP{} token, err = jwt.ParseWithClaims(tokenString, claims, func(t *jwt.Token) (interface{}, error) { if _, ok := t.Method.(*jwt.SigningMethodRSA); !ok { return nil, fmt.Errorf("unexpected signing method: %v", t.Header["alg"]) @@ -219,17 +219,17 @@ func (a *Auth) validateToken(ctx context.Context, tokenString string) (*payloadK } // 4. Validate issuer - //if claims.Issuer != a.issuerURLDocker { - // return nil, fmt.Errorf("invalid issuer: got %s, want %s", claims.Issuer, a.issuerURLDocker) - //} + if claims.Issuer != a.issuerURL { + return nil, fmt.Errorf("invalid issuer: got %s, want %s", claims.Issuer, a.issuerURL) + } return claims, nil } -type payloadKeycloak struct { +type payloadIDP struct { jwt.RegisteredClaims - KeycloakID string `json:"sub"` - SessionID string `json:"sid"` // Keycloak session ID + IDPID string `json:"sub"` + SessionID string `json:"sid"` // IDP session ID Email string `json:"email"` Username string `json:"preferred_username"` ClientName string `json:"azp"` From fd96681ce96e34c8f0956e88aa6185d6d37de98c Mon Sep 17 00:00:00 2001 From: Jan Malte Behrens Date: Mon, 13 Jul 2026 13:09:42 +0200 Subject: [PATCH 14/17] fix: PR Review changes. Environment Variable Update --- auth/auth.go | 75 ++++++++++++++++++++++------------------------------ 1 file changed, 32 insertions(+), 43 deletions(-) diff --git a/auth/auth.go b/auth/auth.go index 90dcf9f7e..77d1c8508 100644 --- a/auth/auth.go +++ b/auth/auth.go @@ -23,9 +23,7 @@ import ( ) var ( - envAuthFake = environment.NewVariable("AUTH_FAKE", "false", "Use user id 1 for every request. Ignores all other auth environment variables.") - - envIssuerURL = environment.NewVariable("IDP_URL_EXTERNAL", "http://localhost:8080", "URL of idp server") + envIssuerURL = environment.NewVariable("IDP_URL_EXTERNAL", "https://localhost:8080", "URL of idp server") envIssuerURLInternal = environment.NewVariable("IDP_URL_INTERNAL", "http://zitadel-api:8080", "Internal URL of idp server") ) @@ -49,14 +47,10 @@ type LogoutEventer interface { // // Has to be initialized with auth.New(). type Auth struct { - fake bool - logedoutSessions *topic.Topic[string] - tokenKey string - cookieKey string - - issuerURLDocker string + issuerURL string + issuerURLInternal string keys map[string]*rsa.PublicKey expiresAt time.Time @@ -67,27 +61,24 @@ type Auth struct { // Returns the initialized Auth objectand a function to be called in the // background. func New(lookup environment.Environmenter, messageBus LogoutEventer) (*Auth, func(context.Context, func(error)), error) { - issuerURLDocker := envIssuerURLInternal.Value(lookup) - if issuerURLDocker == "" { - issuerURLDocker = envIssuerURL.Value(lookup) - } + issuerURL := envIssuerURL.Value(lookup) - fake, _ := strconv.ParseBool(envAuthFake.Value(lookup)) + issuerURLInternal := envIssuerURLInternal.Value(lookup) + if issuerURLInternal == "" { + issuerURLInternal = envIssuerURL.Value(lookup) + } a := &Auth{ - logedoutSessions: topic.New[string](), - issuerURLDocker: issuerURLDocker, - keys: make(map[string]*rsa.PublicKey), + logedoutSessions: topic.New[string](), + issuerURL: issuerURL, + issuerURLInternal: issuerURLInternal, + keys: make(map[string]*rsa.PublicKey), } // Make sure the topic is not empty a.logedoutSessions.Publish("") background := func(ctx context.Context, errorHandler func(error)) { - if fake { - return - } - go a.listenOnLogouts(ctx, messageBus, errorHandler) go a.pruneOldData(ctx) } @@ -98,10 +89,6 @@ func New(lookup environment.Environmenter, messageBus LogoutEventer) (*Auth, fun // Authenticate uses the headers from the given request to get the user id. The // returned context will be cancled, if the session is revoked. func (a *Auth) Authenticate(w http.ResponseWriter, r *http.Request) (context.Context, error) { - if a.fake { - return r.Context(), nil - } - ctx := r.Context() p := new(payloadIDP) @@ -120,8 +107,7 @@ func (a *Auth) Authenticate(w http.ResponseWriter, r *http.Request) (context.Con } // Get OS User Id linked to IDP ID - userID, err := strconv.Atoi(p.OSUserID) - if err != nil { + if userID, err := strconv.Atoi(p.OSUserID); err != nil { return nil, &authError{"user id is not an integer " + p.OSUserID, nil} } @@ -158,22 +144,24 @@ func (a *Auth) loadTokenIDP(w http.ResponseWriter, r *http.Request, payload jwt. return nil } - _, err := a.validateToken(r.Context(), encodedToken) - if err != nil { + if _, err := a.validateToken(r.Context(), encodedToken); err != nil { // OIDC validation failed - return error (no cookie means no legacy fallback) return authError{msg: "Invalid OIDC token", wrapped: err} } - _, err = jwt.ParseWithClaims(encodedToken, payload, func(token *jwt.Token) (interface{}, error) { - kid, ok := token.Header["kid"].(string) + if _, err = jwt.ParseWithClaims(encodedToken, payload, func(token *jwt.Token) (interface{}, error) { + rawKid, ok := token.Header["kid"] if !ok { - return nil, fmt.Errorf("Missing kid in token header") + return nil, fmt.Errorf("missing kid in token header") } - return a.getKey(r.Context(), kid) - }) + kid, ok := rawKid.(string) + if !ok { + return nil, fmt.Errorf("kid in token header is not a string") + } - if err != nil { + return a.getKey(r.Context(), kid) + }; err != nil { var invalid *jwt.ValidationError if errors.As(err, &invalid) { fmt.Println(err) @@ -191,9 +179,14 @@ func (a *Auth) validateToken(ctx context.Context, tokenString string) (*payloadI return nil, fmt.Errorf("parsing token: %w", err) } - kid, ok := token.Header["kid"].(string) + rawKid, ok := token.Header["kid"] if !ok { - return nil, errors.New("missing kid in token header") + return nil, fmt.Errorf("missing kid in token header") + } + + kid, ok := rawKid.(string) + if !ok { + return nil, fmt.Errorf("kid in token header is not a string") } // 2. Get public key from JWKS (cached) @@ -215,7 +208,7 @@ func (a *Auth) validateToken(ctx context.Context, tokenString string) (*payloadI } if !token.Valid { - return nil, errors.New("invalid token") + return nil, fmt.Errorf("invalid token") } // 4. Validate issuer @@ -253,7 +246,7 @@ func (a *Auth) fetchJWKS(ctx context.Context, kid string) (*rsa.PublicKey, error return key, nil } - req, err := http.NewRequestWithContext(ctx, "GET", a.issuerURLDocker+"/oauth/v2/keys", nil) + req, err := http.NewRequestWithContext(ctx, "GET", a.issuerURLInternal+"/oauth/v2/keys", nil) if err != nil { return nil, fmt.Errorf("creating JWKS request: %w", err) } @@ -335,10 +328,6 @@ func (a *Auth) AuthenticatedContext(ctx context.Context, userID int) context.Con // // Panics, if the context was not returned from Authenticate func (a *Auth) FromContext(ctx context.Context) int { - if a.fake { - return 1 - } - v := ctx.Value(userIDType) if v == nil { panic("call to auth.FromContext() without auth.Authenticate()") From 404d6fcce3556f22ee2cfa9c90b3a328c7dd5de7 Mon Sep 17 00:00:00 2001 From: Jan Malte Behrens Date: Mon, 13 Jul 2026 13:27:06 +0200 Subject: [PATCH 15/17] fix: Wrong variable name declarations --- auth/auth.go | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/auth/auth.go b/auth/auth.go index 77d1c8508..83fc27964 100644 --- a/auth/auth.go +++ b/auth/auth.go @@ -107,7 +107,8 @@ func (a *Auth) Authenticate(w http.ResponseWriter, r *http.Request) (context.Con } // Get OS User Id linked to IDP ID - if userID, err := strconv.Atoi(p.OSUserID); err != nil { + userID, err := strconv.Atoi(p.OSUserID) + if err != nil { return nil, &authError{"user id is not an integer " + p.OSUserID, nil} } @@ -149,7 +150,7 @@ func (a *Auth) loadTokenIDP(w http.ResponseWriter, r *http.Request, payload jwt. return authError{msg: "Invalid OIDC token", wrapped: err} } - if _, err = jwt.ParseWithClaims(encodedToken, payload, func(token *jwt.Token) (interface{}, error) { + if _, err := jwt.ParseWithClaims(encodedToken, payload, func(token *jwt.Token) (interface{}, error) { rawKid, ok := token.Header["kid"] if !ok { return nil, fmt.Errorf("missing kid in token header") @@ -161,7 +162,7 @@ func (a *Auth) loadTokenIDP(w http.ResponseWriter, r *http.Request, payload jwt. } return a.getKey(r.Context(), kid) - }; err != nil { + }); err != nil { var invalid *jwt.ValidationError if errors.As(err, &invalid) { fmt.Println(err) From 8b4cd9b3c134ecefe1b893965e0da50b5535fde2 Mon Sep 17 00:00:00 2001 From: Bastian Rihm Date: Fri, 17 Jul 2026 17:38:22 +0200 Subject: [PATCH 16/17] Add throttle package (#303) --- throttle/throttle.go | 72 ++++++++++++++ throttle/throttle_test.go | 203 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 275 insertions(+) create mode 100644 throttle/throttle.go create mode 100644 throttle/throttle_test.go diff --git a/throttle/throttle.go b/throttle/throttle.go new file mode 100644 index 000000000..10242127a --- /dev/null +++ b/throttle/throttle.go @@ -0,0 +1,72 @@ +package throttle + +import ( + "context" + "time" +) + +// Throttler limits how often enqueued functions are executed. +type Throttler struct { + in chan func() +} + +// New returns a new Throttler +func New(ctx context.Context, period time.Duration) *Throttler { + t := &Throttler{ + in: make(chan func(), 1), + } + + go t.loop(ctx, period) + return t +} + +// Run enqueues fn for throttled execution. If a function is already pending +// (waiting for the throttle window to expire), it is replaced by fn so that +// only the most recently enqueued function runs. +func (tt *Throttler) Run(fn func()) { + for { + select { + case tt.in <- fn: + return + default: + } + + select { + case <-tt.in: + default: + } + } +} + +func (tt *Throttler) loop(ctx context.Context, period time.Duration) { + var timer *time.Timer + var timerC <-chan time.Time + var nextFn func() + + for { + select { + case <-ctx.Done(): + if timer != nil { + timer.Stop() + } + return + + case fn := <-tt.in: + if timer == nil { + timer = time.NewTimer(0) + } else { + timer.Reset(period) + } + + nextFn = fn + if timerC == nil { + timerC = timer.C + } + + case <-timerC: + nextFn() + nextFn = nil + timerC = nil + } + } +} diff --git a/throttle/throttle_test.go b/throttle/throttle_test.go new file mode 100644 index 000000000..06b4985fe --- /dev/null +++ b/throttle/throttle_test.go @@ -0,0 +1,203 @@ +package throttle + +import ( + "context" + "sync/atomic" + "testing" + "time" +) + +// TestImmediateExecution verifies that the first Run executes promptly. +func TestImmediateExecution(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + + done := make(chan struct{}) + tt := New(ctx, 50*time.Millisecond) + tt.Run(func() { + close(done) + }) + + select { + case <-done: + case <-time.After(20 * time.Millisecond): + t.Fatal("first Run did not execute instantly") + } +} + +// TestThrottleWindowDelaysSecondRun verifies that a second Run during the +// throttle window is delayed until the window expires. +func TestThrottleWindowDelaysSecondRun(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + + period := 50 * time.Millisecond + tt := New(ctx, period) + + firstDone := make(chan struct{}) + tt.Run(func() { + close(firstDone) + }) + select { + case <-firstDone: + case <-time.After(40 * time.Millisecond): + t.Fatal("first Run did not execute instantly") + } + + start := time.Now() + secondDone := make(chan struct{}) + tt.Run(func() { + close(secondDone) + }) + + select { + case <-secondDone: + elapsed := time.Since(start) + if elapsed < period { + t.Fatalf("second Run executed after %v, expected at least %v", elapsed, period) + } + case <-time.After(2 * time.Second): + t.Fatal("second Run did not execute within 2s") + } +} + +// TestMostRecentWins verifies that when multiple Runs are called during the +// throttle window, only the most recent function executes. +func TestMostRecentWins(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + + period := 50 * time.Millisecond + tt := New(ctx, period) + + firstDone := make(chan struct{}) + tt.Run(func() { + close(firstDone) + }) + <-firstDone + + var executed atomic.Int32 + winnerDone := make(chan struct{}) + + tt.Run(func() { executed.Store(1) }) + tt.Run(func() { executed.Store(2) }) + tt.Run(func() { + executed.Store(3) + close(winnerDone) + }) + + select { + case <-winnerDone: + case <-time.After(2 * time.Second): + t.Fatal("trailing function did not execute within 2s") + } + + if got := executed.Load(); got != 3 { + t.Fatalf("expected last function to win (got %d, want 3)", got) + } +} + +// TestMultiplePeriods verifies that the throttler correctly handles several +// consecutive executions spaced by the period. +func TestMultiplePeriods(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + + period := 30 * time.Millisecond + tt := New(ctx, period) + + var count atomic.Int32 + + for i := range 4 { + expected := int32(i + 1) + notify := make(chan struct{}) + tt.Run(func() { + count.Add(1) + if got := count.Load(); got == expected { + close(notify) + } + }) + + select { + case <-notify: + case <-time.After(2 * time.Second): + t.Fatalf("execution %d did not complete within 2s", i+1) + } + + // Wait for the throttle window to expire before enqueuing the next one. + time.Sleep(period + 10*time.Millisecond) + } + + if got := count.Load(); got != 4 { + t.Fatalf("expected 4 executions, got %d", got) + } +} + +// TestContextCancellation verifies that cancelling the context stops the +// throttler and prevents further executions. +func TestContextCancellation(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + + period := 100 * time.Millisecond + tt := New(ctx, period) + + executed := make(chan struct{}, 1) + tt.Run(func() { + select { + case executed <- struct{}{}: + default: + } + }) + + // Let the first execution complete. + select { + case <-executed: + case <-time.After(time.Second): + t.Fatal("first Run did not execute") + } + + // Cancel the context mid-window. + cancel() + + // Give the goroutine time to observe cancellation. + time.Sleep(period + 50*time.Millisecond) + + // Enqueue another function; it should never execute. + tt.Run(func() { + t.Error("function executed after context cancellation") + }) + + time.Sleep(period + 60*time.Millisecond) +} + +// TestConcurrentRuns verifies that concurrent Run calls don't panic or +// deadlock, and that exactly one function survives to execute. +func TestConcurrentRuns(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + + period := 50 * time.Millisecond + tt := New(ctx, period) + + // Start with one execution to open the throttle window. + ready := make(chan struct{}) + tt.Run(func() { close(ready) }) + <-ready + + // Fire many concurrent Run calls during the window. + var counter atomic.Int32 + for range 100 { + go tt.Run(func() { counter.Add(1) }) + } + + // Wait long enough for the window to expire and the trailing function to run. + time.Sleep(period + 80*time.Millisecond) + + got := counter.Load() + if got == 0 { + t.Fatal("expected at least one trailing execution, got 0") + } + if got > 1 { + t.Logf("note: %d functions executed (expected exactly 1, race may allow 2 in edge cases)", got) + } +} From 175734b733e2de9e864a0fa98c318bbb092107e2 Mon Sep 17 00:00:00 2001 From: Jan Malte Behrens Date: Tue, 21 Jul 2026 15:27:14 +0200 Subject: [PATCH 17/17] feat: Adjust JWT payload, add external Host, fix environment variables, removed outdated blocklisting for now --- auth/auth.go | 66 ++++++++++++++++++++++++++-------------------------- 1 file changed, 33 insertions(+), 33 deletions(-) diff --git a/auth/auth.go b/auth/auth.go index 83fc27964..2f0fb4680 100644 --- a/auth/auth.go +++ b/auth/auth.go @@ -11,8 +11,6 @@ import ( "fmt" "math/big" "net/http" - "slices" - "strconv" "strings" "time" @@ -23,7 +21,8 @@ import ( ) var ( - envIssuerURL = environment.NewVariable("IDP_URL_EXTERNAL", "https://localhost:8080", "URL of idp server") + envExternalHost = environment.NewVariable("IDP_EXTERNAL_HOST", "localhost:8800", "External host address") + envIssuerURL = environment.NewVariable("IDP_URL_EXTERNAL", "https://localhost:8800", "URL of idp server") envIssuerURLInternal = environment.NewVariable("IDP_URL_INTERNAL", "http://zitadel-api:8080", "Internal URL of idp server") ) @@ -49,6 +48,7 @@ type LogoutEventer interface { type Auth struct { logedoutSessions *topic.Topic[string] + externalHost string issuerURL string issuerURLInternal string @@ -61,6 +61,7 @@ type Auth struct { // Returns the initialized Auth objectand a function to be called in the // background. func New(lookup environment.Environmenter, messageBus LogoutEventer) (*Auth, func(context.Context, func(error)), error) { + externalHost := envExternalHost.Value(lookup) issuerURL := envIssuerURL.Value(lookup) issuerURLInternal := envIssuerURLInternal.Value(lookup) @@ -70,6 +71,7 @@ func New(lookup environment.Environmenter, messageBus LogoutEventer) (*Auth, fun a := &Auth{ logedoutSessions: topic.New[string](), + externalHost: externalHost, issuerURL: issuerURL, issuerURLInternal: issuerURLInternal, keys: make(map[string]*rsa.PublicKey), @@ -93,7 +95,6 @@ func (a *Auth) Authenticate(w http.ResponseWriter, r *http.Request) (context.Con p := new(payloadIDP) if err := a.loadTokenIDP(w, r, p); err != nil { - fmt.Println("reading token: %w", err) return nil, fmt.Errorf("reading token: %w", err) } @@ -101,34 +102,31 @@ func (a *Auth) Authenticate(w http.ResponseWriter, r *http.Request) (context.Con return a.AuthenticatedContext(ctx, 0), nil } - cid, sessionIDs := a.logedoutSessions.ReceiveAll() - if slices.Contains(sessionIDs, p.SessionID) { - return nil, &authError{"invalid session", nil} - } + // TODO: Blocklist + //cid, sessionIDs := a.logedoutSessions.ReceiveAll() + //if slices.Contains(sessionIDs, p.SessionID) { + // return nil, &authError{"invalid session", nil} + //} // Get OS User Id linked to IDP ID - userID, err := strconv.Atoi(p.OSUserID) - if err != nil { - return nil, &authError{"user id is not an integer " + p.OSUserID, nil} - } - - ctx, cancelCtx := context.WithCancel(a.AuthenticatedContext(ctx, userID)) + ctx, cancelCtx := context.WithCancel(a.AuthenticatedContext(ctx, p.OSUserID)) go func() { defer cancelCtx() - var sessionIDs []string - var err error - for { - cid, sessionIDs, err = a.logedoutSessions.ReceiveSince(ctx, cid) - if err != nil { - return - } - - if slices.Contains(sessionIDs, p.SessionID) { - return - } - } + /* + var sessionIDs []string + var err error + for { + cid, sessionIDs, err = a.logedoutSessions.ReceiveSince(ctx, cid) + if err != nil { + return + } + + if slices.Contains(sessionIDs, p.SessionID) { + return + } + }*/ }() return ctx, nil @@ -158,6 +156,7 @@ func (a *Auth) loadTokenIDP(w http.ResponseWriter, r *http.Request, payload jwt. kid, ok := rawKid.(string) if !ok { + return nil, fmt.Errorf("kid in token header is not a string") } @@ -165,7 +164,6 @@ func (a *Auth) loadTokenIDP(w http.ResponseWriter, r *http.Request, payload jwt. }); err != nil { var invalid *jwt.ValidationError if errors.As(err, &invalid) { - fmt.Println(err) return a.handleInvalidToken(r.Context(), invalid, w, encodedToken) } } @@ -199,11 +197,13 @@ func (a *Auth) validateToken(ctx context.Context, tokenString string) (*payloadI // 3. Validate token with public key claims := &payloadIDP{} token, err = jwt.ParseWithClaims(tokenString, claims, func(t *jwt.Token) (interface{}, error) { + if _, ok := t.Method.(*jwt.SigningMethodRSA); !ok { return nil, fmt.Errorf("unexpected signing method: %v", t.Header["alg"]) } return key, nil }) + if err != nil { return nil, fmt.Errorf("validating token: %w", err) } @@ -222,12 +222,10 @@ func (a *Auth) validateToken(ctx context.Context, tokenString string) (*payloadI type payloadIDP struct { jwt.RegisteredClaims - IDPID string `json:"sub"` - SessionID string `json:"sid"` // IDP session ID - Email string `json:"email"` - Username string `json:"preferred_username"` - ClientName string `json:"azp"` - OSUserID string `json:"os_id"` + IDPID string `json:"sub"` + Issuer string `json:"iss"` + // SessionID string `json:"sid"` // IDP session ID + OSUserID int `json:"os_id"` } // getKey returns the RSA public key for the given kid, fetching from JWKS if needed @@ -252,6 +250,8 @@ func (a *Auth) fetchJWKS(ctx context.Context, kid string) (*rsa.PublicKey, error return nil, fmt.Errorf("creating JWKS request: %w", err) } + req.Host = a.externalHost + resp, err := http.DefaultClient.Do(req) if err != nil { return nil, fmt.Errorf("fetching JWKS: %w", err)