diff --git a/.gitignore b/.gitignore
index 3093712e..3dae3a42 100644
--- a/.gitignore
+++ b/.gitignore
@@ -43,4 +43,6 @@ nb-configuration.xml
streamx.log
streamx.*
-streamx-*.log
\ No newline at end of file
+streamx-*.log
+# Spec unpacked from the aggregated-openapi artifact at build time (not committed).
+src/main/openapi/
diff --git a/pom.xml b/pom.xml
index 8a528fc4..d4a7b3ed 100644
--- a/pom.xml
+++ b/pom.xml
@@ -23,6 +23,7 @@
3.6.0
25.0.2
2.0.37
+ 2.1.0-SNAPSHOT
4.0.1
false
@@ -52,6 +53,17 @@
quarkus-picocli
+
+ io.quarkiverse.openapi.generator
+ quarkus-openapi-generator
+ 2.11.0
+
+
+
+ io.quarkus
+ quarkus-rest-client-jackson
+
+
io.quarkus
quarkus-arc
@@ -311,6 +323,35 @@
+
+ org.apache.maven.plugins
+ maven-dependency-plugin
+
+
+ unpack-platform-openapi-spec
+ initialize
+
+ unpack
+
+
+
+
+ com.streamx.platform
+ streamx-platform-aggregated-openapi
+ ${aggregated-openapi.version}
+ jar
+ META-INF/openapi/openapi.yaml
+ ${project.basedir}/src/main/openapi
+
+
+
+
+
+ true
+
+
+
+
io.quarkus.platform
quarkus-maven-plugin
diff --git a/src/main/java/com/streamx/cli/auth/AccessTokenClaims.java b/src/main/java/com/streamx/cli/auth/AccessTokenClaims.java
new file mode 100644
index 00000000..ca64f9be
--- /dev/null
+++ b/src/main/java/com/streamx/cli/auth/AccessTokenClaims.java
@@ -0,0 +1,29 @@
+package com.streamx.cli.auth;
+
+import static com.streamx.cli.i18n.MessageProvider.msg;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.streamx.cli.framework.CliException;
+import java.io.IOException;
+import java.util.Base64;
+
+public final class AccessTokenClaims {
+
+ private static final ObjectMapper MAPPER = new ObjectMapper();
+
+ private AccessTokenClaims() {
+ }
+
+ public static JsonNode of(String accessToken) {
+ String[] parts = accessToken.split("\\.");
+ if (parts.length < 2) {
+ throw new CliException(msg.authTokenMalformed());
+ }
+ try {
+ return MAPPER.readTree(Base64.getUrlDecoder().decode(parts[1]));
+ } catch (IOException | IllegalArgumentException e) {
+ throw new CliException(msg.authTokenMalformed(), e);
+ }
+ }
+}
diff --git a/src/main/java/com/streamx/cli/auth/AuthConfig.java b/src/main/java/com/streamx/cli/auth/AuthConfig.java
new file mode 100644
index 00000000..e7bab833
--- /dev/null
+++ b/src/main/java/com/streamx/cli/auth/AuthConfig.java
@@ -0,0 +1,52 @@
+package com.streamx.cli.auth;
+
+import com.streamx.cli.config.StreamxHome;
+import io.smallrye.config.ConfigMapping;
+import io.smallrye.config.PropertiesConfigSource;
+import io.smallrye.config.SmallRyeConfigBuilder;
+import io.smallrye.config.WithDefault;
+import io.smallrye.config.WithName;
+import java.io.IOException;
+import java.util.Optional;
+import org.apache.commons.lang3.BooleanUtils;
+
+@ConfigMapping
+public interface AuthConfig {
+ String DEFAULT_REALM = "streamx";
+ String DEFAULT_CLIENT_ID = "streamx-cli";
+
+ String STREAMX_AUTH_SERVER_URL = "streamx.auth.server-url";
+ String STREAMX_AUTH_REALM = "streamx.auth.realm";
+ String STREAMX_AUTH_CLIENT_ID = "streamx.auth.client-id";
+ String STREAMX_AUTH_INSECURE = "streamx.auth.insecure";
+
+ @WithName(STREAMX_AUTH_SERVER_URL)
+ Optional serverUrl();
+
+ @WithName(STREAMX_AUTH_REALM)
+ @WithDefault(DEFAULT_REALM)
+ String realm();
+
+ @WithName(STREAMX_AUTH_CLIENT_ID)
+ @WithDefault(DEFAULT_CLIENT_ID)
+ String clientId();
+
+ @WithName(STREAMX_AUTH_INSECURE)
+ @WithDefault(BooleanUtils.FALSE)
+ boolean insecure();
+
+ static AuthConfig load() {
+ SmallRyeConfigBuilder builder = new SmallRyeConfigBuilder()
+ .withMapping(AuthConfig.class)
+ .addDefaultSources();
+
+ try {
+ builder.withSources(new PropertiesConfigSource(StreamxHome.getConfigUrl(), 260));
+ } catch (IOException expected) {
+ }
+
+ return builder
+ .build()
+ .getConfigMapping(AuthConfig.class);
+ }
+}
diff --git a/src/main/java/com/streamx/cli/auth/Credentials.java b/src/main/java/com/streamx/cli/auth/Credentials.java
new file mode 100644
index 00000000..2bda0716
--- /dev/null
+++ b/src/main/java/com/streamx/cli/auth/Credentials.java
@@ -0,0 +1,13 @@
+package com.streamx.cli.auth;
+
+import java.time.Instant;
+
+public record Credentials(
+ String accessToken,
+ String refreshToken,
+ Instant expiresAt,
+ String issuerUrl,
+ String clientId,
+ boolean insecure
+) {
+}
diff --git a/src/main/java/com/streamx/cli/auth/CredentialsStore.java b/src/main/java/com/streamx/cli/auth/CredentialsStore.java
new file mode 100644
index 00000000..45c480e6
--- /dev/null
+++ b/src/main/java/com/streamx/cli/auth/CredentialsStore.java
@@ -0,0 +1,119 @@
+package com.streamx.cli.auth;
+
+import static com.streamx.cli.i18n.MessageProvider.msg;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.streamx.cli.config.StreamxHome;
+import com.streamx.cli.framework.CliException;
+import java.io.IOException;
+import java.nio.file.AtomicMoveNotSupportedException;
+import java.nio.file.FileSystems;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.StandardCopyOption;
+import java.nio.file.attribute.PosixFilePermission;
+import java.nio.file.attribute.PosixFilePermissions;
+import java.time.Instant;
+import java.util.Optional;
+import java.util.Set;
+
+public class CredentialsStore {
+ private static final String CREDENTIALS_FILE = "credentials.json";
+ private static final ObjectMapper MAPPER = new ObjectMapper();
+
+ private static final String ACCESS_TOKEN = "access_token";
+ private static final String REFRESH_TOKEN = "refresh_token";
+ private static final String EXPIRES_AT = "expires_at";
+ private static final String ISSUER_URL = "issuer_url";
+ private static final String CLIENT_ID = "client_id";
+ private static final String INSECURE = "insecure";
+
+ public static Path getCredentialsPath() {
+ return StreamxHome.getConfigDir().resolve(CREDENTIALS_FILE);
+ }
+
+ public static boolean exists() {
+ return Files.isRegularFile(getCredentialsPath());
+ }
+
+ public static Optional load() {
+ Path path = getCredentialsPath();
+ if (!Files.isRegularFile(path)) {
+ return Optional.empty();
+ }
+
+ try {
+ JsonNode node = MAPPER.readTree(Files.readString(path));
+ return Optional.of(new Credentials(
+ node.path(ACCESS_TOKEN).asText(null),
+ node.path(REFRESH_TOKEN).asText(null),
+ Instant.ofEpochSecond(node.path(EXPIRES_AT).asLong()),
+ node.path(ISSUER_URL).asText(null),
+ node.path(CLIENT_ID).asText(null),
+ node.path(INSECURE).asBoolean(false)
+ ));
+ } catch (IOException e) {
+ throw new CliException(msg.authCredentialsUnreadable(path.toString(), e.getMessage()), e);
+ }
+ }
+
+ public static void save(Credentials credentials) {
+ Path path = getCredentialsPath();
+
+ ObjectNode node = MAPPER.createObjectNode();
+ node.put(ACCESS_TOKEN, credentials.accessToken());
+ node.put(REFRESH_TOKEN, credentials.refreshToken());
+ node.put(EXPIRES_AT, credentials.expiresAt().getEpochSecond());
+ node.put(ISSUER_URL, credentials.issuerUrl());
+ node.put(CLIENT_ID, credentials.clientId());
+ node.put(INSECURE, credentials.insecure());
+
+ Path temporary = path.resolveSibling(path.getFileName() + ".tmp");
+ try {
+ Files.createDirectories(path.getParent());
+ Files.deleteIfExists(temporary);
+ createOwnerOnlyFile(temporary);
+ Files.writeString(temporary, MAPPER.writeValueAsString(node));
+ moveIntoPlace(temporary, path);
+ } catch (IOException e) {
+ quietlyDelete(temporary);
+ throw new CliException(msg.authCredentialsNotSaved(path.toString(), e.getMessage()), e);
+ }
+ }
+
+ private static void moveIntoPlace(Path temporary, Path path) throws IOException {
+ try {
+ Files.move(temporary, path, StandardCopyOption.REPLACE_EXISTING,
+ StandardCopyOption.ATOMIC_MOVE);
+ } catch (AtomicMoveNotSupportedException e) {
+ Files.move(temporary, path, StandardCopyOption.REPLACE_EXISTING);
+ }
+ }
+
+ private static void quietlyDelete(Path path) {
+ try {
+ Files.deleteIfExists(path);
+ } catch (IOException expected) {
+ }
+ }
+
+ public static void delete() {
+ Path path = getCredentialsPath();
+ try {
+ Files.deleteIfExists(path);
+ } catch (IOException e) {
+ throw new CliException(msg.authCredentialsNotDeleted(path.toString(), e.getMessage()), e);
+ }
+ }
+
+ private static void createOwnerOnlyFile(Path path) throws IOException {
+ if (FileSystems.getDefault().supportedFileAttributeViews().contains("posix")) {
+ Files.createFile(path, PosixFilePermissions.asFileAttribute(
+ Set.of(PosixFilePermission.OWNER_READ, PosixFilePermission.OWNER_WRITE)));
+ } else {
+ Files.createFile(path);
+ }
+ }
+}
diff --git a/src/main/java/com/streamx/cli/auth/Identity.java b/src/main/java/com/streamx/cli/auth/Identity.java
new file mode 100644
index 00000000..2a3a3b3b
--- /dev/null
+++ b/src/main/java/com/streamx/cli/auth/Identity.java
@@ -0,0 +1,16 @@
+package com.streamx.cli.auth;
+
+import io.quarkus.runtime.annotations.RegisterForReflection;
+
+@RegisterForReflection
+public record Identity(
+ String username,
+ String name,
+ String email,
+ String subject,
+ String issuer,
+ String expiresAt,
+ boolean expired,
+ String tokenId
+) {
+}
diff --git a/src/main/java/com/streamx/cli/auth/InteractiveGrant.java b/src/main/java/com/streamx/cli/auth/InteractiveGrant.java
new file mode 100644
index 00000000..01d819b3
--- /dev/null
+++ b/src/main/java/com/streamx/cli/auth/InteractiveGrant.java
@@ -0,0 +1,58 @@
+package com.streamx.cli.auth;
+
+import static com.streamx.cli.i18n.MessageProvider.msg;
+
+import com.streamx.cli.auth.OidcClient.Endpoints;
+import com.streamx.cli.auth.OidcDeviceFlow.DeviceAuthorization;
+import com.streamx.cli.framework.CliException;
+
+public final class InteractiveGrant {
+
+ private InteractiveGrant() {
+ }
+
+ public static OidcClient clientFromConfig(AuthConfig config) {
+ String serverUrl = config.serverUrl()
+ .filter(url -> !url.isBlank())
+ .orElseThrow(() -> new CliException(
+ msg.authServerUrlNotConfigured(AuthConfig.STREAMX_AUTH_SERVER_URL)));
+
+ return new OidcClient(
+ OidcClient.issuerUrl(serverUrl, config.realm()),
+ config.clientId(),
+ config.insecure());
+ }
+
+ public static boolean preferDeviceFlow(boolean noBrowser) {
+ return noBrowser
+ || System.getenv("SSH_CONNECTION") != null
+ || System.getenv("SSH_TTY") != null;
+ }
+
+ public static Credentials run(OidcClient client, Endpoints endpoints, String scope,
+ boolean noBrowser) {
+ if (preferDeviceFlow(noBrowser)) {
+ return deviceGrant(client, endpoints, scope);
+ }
+ try {
+ return new OidcAuthCodeFlow(client).login(endpoints, scope);
+ } catch (OidcAuthCodeFlow.BrowserUnavailableException e) {
+ System.err.println(msg.authBrowserFallbackToDevice());
+ return deviceGrant(client, endpoints, scope);
+ }
+ }
+
+ private static Credentials deviceGrant(OidcClient client, Endpoints endpoints, String scope) {
+ OidcDeviceFlow flow = new OidcDeviceFlow(client);
+ DeviceAuthorization authorization = flow.requestDeviceAuthorization(endpoints, scope);
+
+ System.err.println(msg.authLoginInstructions(
+ authorization.verificationUri(),
+ authorization.userCode()));
+ if (authorization.verificationUriComplete() != null) {
+ System.err.println(msg.authLoginDirectLink(authorization.verificationUriComplete()));
+ }
+
+ return flow.pollForToken(endpoints, authorization);
+ }
+}
diff --git a/src/main/java/com/streamx/cli/auth/OidcAuthCodeFlow.java b/src/main/java/com/streamx/cli/auth/OidcAuthCodeFlow.java
new file mode 100644
index 00000000..4e672522
--- /dev/null
+++ b/src/main/java/com/streamx/cli/auth/OidcAuthCodeFlow.java
@@ -0,0 +1,238 @@
+package com.streamx.cli.auth;
+
+import static com.streamx.cli.i18n.MessageProvider.msg;
+
+import com.streamx.cli.auth.OidcClient.Endpoints;
+import com.streamx.cli.auth.OidcClient.Response;
+import com.streamx.cli.framework.CliException;
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpServer;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.InetSocketAddress;
+import java.net.URI;
+import java.net.URLEncoder;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.Map;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+public class OidcAuthCodeFlow {
+ private static final String LOOPBACK_HOST = "127.0.0.1";
+ private static final String CALLBACK_PATH = "/callback";
+ private static final Duration LOGIN_TIMEOUT = Duration.ofMinutes(5);
+
+ private final OidcClient client;
+ private final BrowserLauncher browserLauncher;
+
+ public OidcAuthCodeFlow(OidcClient client) {
+ this(client, OidcAuthCodeFlow::openInBrowser);
+ }
+
+ public OidcAuthCodeFlow(OidcClient client, BrowserLauncher browserLauncher) {
+ this.client = client;
+ this.browserLauncher = browserLauncher;
+ }
+
+ public interface BrowserLauncher {
+ void open(String url) throws IOException;
+ }
+
+ public Credentials login(Endpoints endpoints) {
+ return login(endpoints, Scopes.LOGIN);
+ }
+
+ public Credentials login(Endpoints endpoints, String scope) {
+ if (endpoints.authorizationEndpoint() == null) {
+ throw new CliException(msg.authCodeFlowUnsupported());
+ }
+
+ Pkce pkce = Pkce.generate();
+ String state = Pkce.randomUrlSafe(32);
+
+ LoopbackReceiver receiver = new LoopbackReceiver(state);
+ receiver.start();
+ try {
+ String authUrl = authorizationUrl(endpoints, receiver.redirectUri(), pkce, state, scope);
+ try {
+ browserLauncher.open(authUrl);
+ } catch (IOException e) {
+ throw new BrowserUnavailableException();
+ }
+
+ System.err.println(msg.authLoginOpeningBrowser());
+ System.err.println(" " + authUrl);
+
+ String code = receiver.awaitCode();
+ return exchangeCode(endpoints, code, pkce.verifier(), receiver.redirectUri());
+ } finally {
+ receiver.stop();
+ }
+ }
+
+ public static class BrowserUnavailableException extends RuntimeException {
+ }
+
+ private String authorizationUrl(
+ Endpoints endpoints, String redirectUri, Pkce pkce, String state, String scope) {
+ Map params = Map.of(
+ "client_id", client.clientId(),
+ "response_type", "code",
+ "scope", scope,
+ "redirect_uri", redirectUri,
+ "state", state,
+ "code_challenge", pkce.challenge(),
+ "code_challenge_method", Pkce.METHOD
+ );
+ StringBuilder query = new StringBuilder();
+ for (Map.Entry entry : params.entrySet()) {
+ if (!query.isEmpty()) {
+ query.append('&');
+ }
+ query.append(entry.getKey()).append('=')
+ .append(URLEncoder.encode(entry.getValue(), StandardCharsets.UTF_8));
+ }
+ return endpoints.authorizationEndpoint() + "?" + query;
+ }
+
+ private Credentials exchangeCode(
+ Endpoints endpoints, String code, String codeVerifier, String redirectUri) {
+ String url = endpoints.tokenEndpoint();
+ Response response = client.exchange(OidcClient.postForm(url, Map.of(
+ "grant_type", "authorization_code",
+ "code", code,
+ "client_id", client.clientId(),
+ "redirect_uri", redirectUri,
+ "code_verifier", codeVerifier
+ )), url);
+
+ if (!response.isSuccess()) {
+ String error = response.body().path("error").asText(null);
+ throw new CliException(
+ error != null ? msg.authLoginFailed(error)
+ : msg.authRequestFailedWithStatus(url, response.statusCode()));
+ }
+ return client.toCredentials(response.body(), null);
+ }
+
+ private static void openInBrowser(String url) throws IOException {
+ String os = System.getProperty("os.name", "").toLowerCase();
+ ProcessBuilder builder;
+ if (os.contains("mac")) {
+ builder = new ProcessBuilder("open", url);
+ } else if (os.contains("win")) {
+ builder = new ProcessBuilder("rundll32", "url.dll,FileProtocolHandler", url);
+ } else {
+ builder = new ProcessBuilder("xdg-open", url);
+ }
+ builder.redirectOutput(ProcessBuilder.Redirect.DISCARD)
+ .redirectError(ProcessBuilder.Redirect.DISCARD)
+ .start();
+ }
+
+ private static final class LoopbackReceiver {
+ private final String expectedState;
+ private final BlockingQueue result = new ArrayBlockingQueue<>(1);
+ private HttpServer server;
+
+ private LoopbackReceiver(String expectedState) {
+ this.expectedState = expectedState;
+ }
+
+ private record Result(String code, String error) {
+ }
+
+ private void start() {
+ try {
+ server = HttpServer.create(new InetSocketAddress(LOOPBACK_HOST, 0), 0);
+ } catch (IOException e) {
+ throw new CliException(msg.authLoopbackFailed(e.getMessage()), e);
+ }
+ server.createContext(CALLBACK_PATH, this::handle);
+ server.start();
+ }
+
+ private String redirectUri() {
+ return "http://" + LOOPBACK_HOST + ":" + server.getAddress().getPort() + CALLBACK_PATH;
+ }
+
+ private void handle(HttpExchange exchange) throws IOException {
+ Map query = parseQuery(exchange.getRequestURI());
+ String state = query.get("state");
+
+ // Validate state before anything else (RFC 6749 ยง4.1.2.1), on both success and error.
+ if (!expectedState.equals(state)) {
+ writeHtml(exchange, msg.authLoopbackDenied());
+ return;
+ }
+
+ String error = query.get("error");
+ String code = query.get("code");
+ String body;
+ if (error != null) {
+ body = msg.authLoopbackDenied();
+ result.offer(new Result(null, error));
+ } else if (code != null) {
+ body = msg.authLoopbackSuccess();
+ result.offer(new Result(code, null));
+ } else {
+ body = msg.authLoopbackDenied();
+ result.offer(new Result(null, "invalid_request"));
+ }
+ writeHtml(exchange, body);
+ }
+
+ private static void writeHtml(HttpExchange exchange, String body) throws IOException {
+ byte[] bytes = ("" + body + "").getBytes(StandardCharsets.UTF_8);
+ exchange.getResponseHeaders().add("Content-Type", "text/html; charset=utf-8");
+ exchange.sendResponseHeaders(200, bytes.length);
+ try (OutputStream os = exchange.getResponseBody()) {
+ os.write(bytes);
+ }
+ }
+
+ private String awaitCode() {
+ Result received;
+ try {
+ received = result.poll(LOGIN_TIMEOUT.toSeconds(), TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new CliException(msg.authLoginInterrupted(), e);
+ }
+ if (received == null) {
+ throw new CliException(msg.authLoginExpired());
+ }
+ if (received.error() != null) {
+ throw new CliException("access_denied".equals(received.error())
+ ? msg.authLoginDenied()
+ : msg.authLoginFailed(received.error()));
+ }
+ return received.code();
+ }
+
+ private void stop() {
+ if (server != null) {
+ server.stop(0);
+ }
+ }
+
+ private static Map parseQuery(URI uri) {
+ Map params = new java.util.HashMap<>();
+ String query = uri.getRawQuery();
+ if (query == null) {
+ return params;
+ }
+ for (String pair : query.split("&")) {
+ int eq = pair.indexOf('=');
+ if (eq > 0) {
+ params.put(
+ java.net.URLDecoder.decode(pair.substring(0, eq), StandardCharsets.UTF_8),
+ java.net.URLDecoder.decode(pair.substring(eq + 1), StandardCharsets.UTF_8));
+ }
+ }
+ return params;
+ }
+ }
+}
diff --git a/src/main/java/com/streamx/cli/auth/OidcClient.java b/src/main/java/com/streamx/cli/auth/OidcClient.java
new file mode 100644
index 00000000..9e13ce92
--- /dev/null
+++ b/src/main/java/com/streamx/cli/auth/OidcClient.java
@@ -0,0 +1,245 @@
+package com.streamx.cli.auth;
+
+import static com.streamx.cli.i18n.MessageProvider.msg;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.streamx.cli.framework.CliException;
+import com.streamx.cli.framework.Urls;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.security.GeneralSecurityException;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import javax.net.ssl.SSLContext;
+import org.apache.http.NameValuePair;
+import org.apache.http.client.config.RequestConfig;
+import org.apache.http.client.entity.UrlEncodedFormEntity;
+import org.apache.http.client.methods.CloseableHttpResponse;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.client.methods.HttpPost;
+import org.apache.http.client.methods.HttpUriRequest;
+import org.apache.http.conn.ssl.NoopHostnameVerifier;
+import org.apache.http.conn.ssl.TrustAllStrategy;
+import org.apache.http.impl.client.CloseableHttpClient;
+import org.apache.http.impl.client.HttpClients;
+import org.apache.http.message.BasicNameValuePair;
+import org.apache.http.ssl.SSLContexts;
+import org.apache.http.util.EntityUtils;
+
+public class OidcClient {
+ private static final String DISCOVERY_PATH = "/.well-known/openid-configuration";
+ static final Duration REQUEST_TIMEOUT = Duration.ofSeconds(30);
+
+ private static final ObjectMapper MAPPER = new ObjectMapper();
+
+ private final String issuerUrl;
+ private final String clientId;
+ private final boolean insecure;
+ private final CloseableHttpClient httpClient;
+
+ public OidcClient(String issuerUrl, String clientId, boolean insecure) {
+ if (Urls.isCleartextRemote(issuerUrl)) {
+ throw new CliException(msg.authCleartextHttpBlocked(issuerUrl));
+ }
+ this.issuerUrl = issuerUrl;
+ this.clientId = clientId;
+ this.insecure = insecure;
+ this.httpClient = buildHttpClient(insecure);
+ }
+
+ public static String issuerUrl(String serverUrl, String realm) {
+ String base = serverUrl.endsWith("/")
+ ? serverUrl.substring(0, serverUrl.length() - 1)
+ : serverUrl;
+ return base + "/realms/" + realm;
+ }
+
+ public String clientId() {
+ return clientId;
+ }
+
+ public record Endpoints(
+ String authorizationEndpoint,
+ String deviceAuthorizationEndpoint,
+ String tokenEndpoint,
+ String revocationEndpoint
+ ) {
+ }
+
+ public Endpoints discover() {
+ String url = issuerUrl + DISCOVERY_PATH;
+ JsonNode node = send(get(url), url, true);
+
+ String documentIssuer = node.path("issuer").asText(null);
+ if (!issuerUrl.equals(documentIssuer)) {
+ throw new CliException(msg.authIssuerMismatch(issuerUrl, String.valueOf(documentIssuer)));
+ }
+ String authorization = node.path("authorization_endpoint").asText(null);
+ String device = node.path("device_authorization_endpoint").asText(null);
+ String token = node.path("token_endpoint").asText(null);
+ String revocation = node.path("revocation_endpoint").asText(null);
+
+ for (String endpoint : new String[] {authorization, device, token, revocation}) {
+ if (endpoint != null && Urls.isCleartextRemote(endpoint)) {
+ throw new CliException(msg.authCleartextHttpBlocked(endpoint));
+ }
+ }
+ return new Endpoints(
+ authorization,
+ device,
+ token,
+ revocation
+ );
+ }
+
+ public Credentials refresh(String refreshToken) {
+ String url = discover().tokenEndpoint();
+
+ Response response = exchange(postForm(url, Map.of(
+ "grant_type", "refresh_token",
+ "refresh_token", refreshToken,
+ "client_id", clientId
+ )), url);
+
+ if (!response.isSuccess()) {
+ String error = response.body().path("error").asText(null);
+ // Only invalid_grant means the refresh token is dead; other errors keep the session.
+ if ("invalid_grant".equals(error)) {
+ throw new CliException(msg.authSessionExpired());
+ }
+ if (error != null) {
+ String detail = response.body().path("error_description").asText(error);
+ throw new CliException(msg.authTokenRequestRejected(response.statusCode(), detail));
+ }
+ throw new CliException(msg.authRequestFailedWithStatus(url, response.statusCode()));
+ }
+ return toCredentials(response.body(), refreshToken);
+ }
+
+ public void revoke(String refreshToken) {
+ Endpoints endpoints = discover();
+ String url = endpoints.revocationEndpoint();
+ if (url == null) {
+ throw new CliException(msg.authRevocationUnsupported());
+ }
+ send(postForm(url, Map.of(
+ "client_id", clientId,
+ "token", refreshToken,
+ "token_type_hint", "refresh_token"
+ )), url, true);
+ }
+
+ public void revokeQuietly(String refreshToken) {
+ try {
+ Endpoints endpoints = discover();
+ if (endpoints.revocationEndpoint() == null) {
+ return;
+ }
+ send(postForm(endpoints.revocationEndpoint(), Map.of(
+ "client_id", clientId,
+ "token", refreshToken,
+ "token_type_hint", "refresh_token"
+ )), endpoints.revocationEndpoint(), false);
+ } catch (RuntimeException expected) {
+ }
+ }
+
+ Credentials toCredentials(JsonNode node, String fallbackRefreshToken) {
+ String accessToken = node.path("access_token").asText(null);
+ if (accessToken == null || accessToken.isBlank()) {
+ throw new CliException(msg.authTokenResponseIncomplete());
+ }
+ return new Credentials(
+ accessToken,
+ node.path("refresh_token").asText(fallbackRefreshToken),
+ Instant.now().plusSeconds(node.path("expires_in").asLong(0)),
+ issuerUrl,
+ clientId,
+ insecure
+ );
+ }
+
+ static HttpGet get(String url) {
+ HttpGet request = new HttpGet(url);
+ request.setHeader("Accept", "application/json");
+ return request;
+ }
+
+ static HttpPost postForm(String url, Map form) {
+ HttpPost request = new HttpPost(url);
+ request.setHeader("Accept", "application/json");
+
+ List params = new ArrayList<>();
+ for (Map.Entry entry : form.entrySet()) {
+ params.add(new BasicNameValuePair(entry.getKey(), entry.getValue()));
+ }
+ request.setEntity(new UrlEncodedFormEntity(params, StandardCharsets.UTF_8));
+ return request;
+ }
+
+ record Response(int statusCode, JsonNode body) {
+ boolean isSuccess() {
+ return statusCode >= 200 && statusCode < 300;
+ }
+ }
+
+ JsonNode send(HttpUriRequest request, String url, boolean failOnErrorStatus) {
+ Response response = exchange(request, url);
+ if (failOnErrorStatus && !response.isSuccess()) {
+ throw new CliException(msg.authRequestFailedWithStatus(url, response.statusCode()));
+ }
+ return response.body();
+ }
+
+ Response exchange(HttpUriRequest request, String url) {
+ String body;
+ int statusCode;
+ try (CloseableHttpResponse response = httpClient.execute(request)) {
+ statusCode = response.getStatusLine().getStatusCode();
+ body = response.getEntity() == null
+ ? ""
+ : EntityUtils.toString(response.getEntity(), StandardCharsets.UTF_8);
+ } catch (IOException e) {
+ throw new CliException(msg.authRequestFailed(url, e.getMessage()), e);
+ }
+
+ try {
+ JsonNode parsed = body.isBlank() ? MAPPER.createObjectNode() : MAPPER.readTree(body);
+ return new Response(statusCode, parsed);
+ } catch (IOException e) {
+ throw new CliException(msg.authResponseNotJson(url), e);
+ }
+ }
+
+ private static CloseableHttpClient buildHttpClient(boolean insecure) {
+ int timeoutMillis = (int) REQUEST_TIMEOUT.toMillis();
+ RequestConfig requestConfig = RequestConfig.custom()
+ .setConnectTimeout(timeoutMillis)
+ .setConnectionRequestTimeout(timeoutMillis)
+ .setSocketTimeout(timeoutMillis)
+ .build();
+
+ if (!insecure) {
+ return HttpClients.custom()
+ .setDefaultRequestConfig(requestConfig)
+ .build();
+ }
+
+ try {
+ SSLContext sslContext = SSLContexts.custom()
+ .loadTrustMaterial(null, TrustAllStrategy.INSTANCE)
+ .build();
+ return HttpClients.custom()
+ .setDefaultRequestConfig(requestConfig)
+ .setSSLContext(sslContext)
+ .setSSLHostnameVerifier(NoopHostnameVerifier.INSTANCE)
+ .build();
+ } catch (GeneralSecurityException e) {
+ throw new CliException(msg.authInsecureTlsFailed(e.getMessage()), e);
+ }
+ }
+}
diff --git a/src/main/java/com/streamx/cli/auth/OidcDeviceFlow.java b/src/main/java/com/streamx/cli/auth/OidcDeviceFlow.java
new file mode 100644
index 00000000..db2a8754
--- /dev/null
+++ b/src/main/java/com/streamx/cli/auth/OidcDeviceFlow.java
@@ -0,0 +1,113 @@
+package com.streamx.cli.auth;
+
+import static com.streamx.cli.i18n.MessageProvider.msg;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.streamx.cli.auth.OidcClient.Endpoints;
+import com.streamx.cli.auth.OidcClient.Response;
+import com.streamx.cli.framework.CliException;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.Map;
+
+public class OidcDeviceFlow {
+ private static final String DEVICE_CODE_GRANT = "urn:ietf:params:oauth:grant-type:device_code";
+ private static final int DEFAULT_POLL_INTERVAL_SECONDS = 5;
+
+ private static final int SLOW_DOWN_INCREMENT_SECONDS = 5;
+
+ private final OidcClient client;
+
+ public OidcDeviceFlow(OidcClient client) {
+ this.client = client;
+ }
+
+ public record DeviceAuthorization(
+ String deviceCode,
+ String userCode,
+ String verificationUri,
+ String verificationUriComplete,
+ int intervalSeconds,
+ Instant expiresAt,
+ Pkce pkce
+ ) {
+ }
+
+ public DeviceAuthorization requestDeviceAuthorization(Endpoints endpoints) {
+ return requestDeviceAuthorization(endpoints, Scopes.LOGIN);
+ }
+
+ public DeviceAuthorization requestDeviceAuthorization(Endpoints endpoints, String scope) {
+ String url = endpoints.deviceAuthorizationEndpoint();
+ if (url == null) {
+ throw new CliException(msg.authDeviceFlowUnsupported(url));
+ }
+ Pkce pkce = Pkce.generate();
+ Map form = Map.of(
+ "client_id", client.clientId(),
+ "scope", scope,
+ "code_challenge", pkce.challenge(),
+ "code_challenge_method", Pkce.METHOD);
+ JsonNode node = client.send(OidcClient.postForm(url, form), url, true);
+
+ int interval = node.path("interval").asInt(DEFAULT_POLL_INTERVAL_SECONDS);
+ long expiresIn = node.path("expires_in").asLong(0);
+
+ return new DeviceAuthorization(
+ node.path("device_code").asText(null),
+ node.path("user_code").asText(null),
+ node.path("verification_uri").asText(null),
+ node.path("verification_uri_complete").asText(null),
+ interval > 0 ? interval : DEFAULT_POLL_INTERVAL_SECONDS,
+ Instant.now().plusSeconds(expiresIn),
+ pkce
+ );
+ }
+
+ public Credentials pollForToken(Endpoints endpoints, DeviceAuthorization authorization) {
+ String url = endpoints.tokenEndpoint();
+ Map form = Map.of(
+ "grant_type", DEVICE_CODE_GRANT,
+ "device_code", authorization.deviceCode(),
+ "client_id", client.clientId(),
+ "code_verifier", authorization.pkce().verifier()
+ );
+
+ int intervalSeconds = authorization.intervalSeconds();
+
+ while (Instant.now().isBefore(authorization.expiresAt())) {
+ sleepSeconds(intervalSeconds);
+
+ Response response = client.exchange(OidcClient.postForm(url, form), url);
+ String error = response.body().path("error").asText(null);
+
+ if (response.isSuccess()) {
+ return client.toCredentials(response.body(), null);
+ }
+ if (error == null) {
+ throw new CliException(msg.authRequestFailedWithStatus(url, response.statusCode()));
+ }
+
+ if ("slow_down".equals(error)) {
+ intervalSeconds += SLOW_DOWN_INCREMENT_SECONDS;
+ } else if ("access_denied".equals(error)) {
+ throw new CliException(msg.authLoginDenied());
+ } else if ("expired_token".equals(error)) {
+ throw new CliException(msg.authLoginExpired());
+ } else if (!"authorization_pending".equals(error)) {
+ throw new CliException(msg.authLoginFailed(error));
+ }
+ }
+
+ throw new CliException(msg.authLoginExpired());
+ }
+
+ private void sleepSeconds(int seconds) {
+ try {
+ Thread.sleep(Duration.ofSeconds(seconds).toMillis());
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new CliException(msg.authLoginInterrupted(), e);
+ }
+ }
+}
diff --git a/src/main/java/com/streamx/cli/auth/Pkce.java b/src/main/java/com/streamx/cli/auth/Pkce.java
new file mode 100644
index 00000000..ebc17aa5
--- /dev/null
+++ b/src/main/java/com/streamx/cli/auth/Pkce.java
@@ -0,0 +1,37 @@
+package com.streamx.cli.auth;
+
+import static com.streamx.cli.i18n.MessageProvider.msg;
+
+import com.streamx.cli.framework.CliException;
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.security.SecureRandom;
+import java.util.Base64;
+
+public record Pkce(String verifier, String challenge) {
+
+ public static final String METHOD = "S256";
+
+ public static Pkce generate() {
+ String verifier = randomUrlSafe(64);
+ return new Pkce(verifier, challenge(verifier));
+ }
+
+ public static String randomUrlSafe(int bytes) {
+ byte[] buffer = new byte[bytes];
+ // Per-call, never a static field: GraalVM rejects a SecureRandom in the native image heap.
+ new SecureRandom().nextBytes(buffer);
+ return Base64.getUrlEncoder().withoutPadding().encodeToString(buffer);
+ }
+
+ private static String challenge(String verifier) {
+ try {
+ byte[] digest = MessageDigest.getInstance("SHA-256")
+ .digest(verifier.getBytes(StandardCharsets.US_ASCII));
+ return Base64.getUrlEncoder().withoutPadding().encodeToString(digest);
+ } catch (NoSuchAlgorithmException e) {
+ throw new CliException(msg.authPkceFailed(e.getMessage()), e);
+ }
+ }
+}
diff --git a/src/main/java/com/streamx/cli/auth/Scopes.java b/src/main/java/com/streamx/cli/auth/Scopes.java
new file mode 100644
index 00000000..f31acf9e
--- /dev/null
+++ b/src/main/java/com/streamx/cli/auth/Scopes.java
@@ -0,0 +1,9 @@
+package com.streamx.cli.auth;
+
+public final class Scopes {
+
+ public static final String LOGIN = "openid profile email";
+
+ private Scopes() {
+ }
+}
diff --git a/src/main/java/com/streamx/cli/commands/StreamxCommand.java b/src/main/java/com/streamx/cli/commands/StreamxCommand.java
index 808126c1..49c89dbf 100644
--- a/src/main/java/com/streamx/cli/commands/StreamxCommand.java
+++ b/src/main/java/com/streamx/cli/commands/StreamxCommand.java
@@ -1,5 +1,6 @@
package com.streamx.cli.commands;
+import com.streamx.cli.commands.auth.AuthCommand;
import com.streamx.cli.commands.completion.CompleteContextNamesCommand;
import com.streamx.cli.commands.completion.CompleteNonDefaultTemplateIdsCommand;
import com.streamx.cli.commands.completion.CompleteRegisteredTemplateIdsCommand;
@@ -8,6 +9,7 @@
import com.streamx.cli.commands.completion.CompleteTemplateIdsCommand;
import com.streamx.cli.commands.completion.CompletionCommand;
import com.streamx.cli.commands.context.ContextCommand;
+import com.streamx.cli.commands.info.InfoCommand;
import com.streamx.cli.commands.local.LocalCommand;
import com.streamx.cli.commands.publish.PublishCommand;
import com.streamx.cli.commands.settings.SettingsCommand;
@@ -18,10 +20,12 @@
name = "streamx",
header = "StreamX CLI. More info at https://streamx.com",
subcommands = {
+ AuthCommand.class,
ContextCommand.class,
LocalCommand.class,
SettingsCommand.class,
PublishCommand.class,
+ InfoCommand.class,
CompletionCommand.class,
CompleteTemplateIdsCommand.class,
CompleteRegisteredTemplateIdsCommand.class,
diff --git a/src/main/java/com/streamx/cli/commands/auth/AuthCommand.java b/src/main/java/com/streamx/cli/commands/auth/AuthCommand.java
new file mode 100644
index 00000000..e4d2cb8b
--- /dev/null
+++ b/src/main/java/com/streamx/cli/commands/auth/AuthCommand.java
@@ -0,0 +1,19 @@
+package com.streamx.cli.commands.auth;
+
+import com.streamx.cli.commands.auth.login.LoginCommand;
+import com.streamx.cli.commands.auth.logout.LogoutCommand;
+import com.streamx.cli.commands.auth.whoami.WhoamiCommand;
+import com.streamx.cli.framework.AbstractCommandGroup;
+import picocli.CommandLine;
+
+@CommandLine.Command(
+ name = "auth",
+ header = "Manage StreamX authentication",
+ subcommands = {
+ LoginCommand.class,
+ LogoutCommand.class,
+ WhoamiCommand.class
+ }
+)
+public class AuthCommand extends AbstractCommandGroup {
+}
diff --git a/src/main/java/com/streamx/cli/commands/auth/login/LoginCommand.java b/src/main/java/com/streamx/cli/commands/auth/login/LoginCommand.java
new file mode 100644
index 00000000..ba87d6d9
--- /dev/null
+++ b/src/main/java/com/streamx/cli/commands/auth/login/LoginCommand.java
@@ -0,0 +1,59 @@
+package com.streamx.cli.commands.auth.login;
+
+import static com.streamx.cli.i18n.MessageProvider.msg;
+
+import com.streamx.cli.auth.AuthConfig;
+import com.streamx.cli.auth.Credentials;
+import com.streamx.cli.auth.CredentialsStore;
+import com.streamx.cli.auth.InteractiveGrant;
+import com.streamx.cli.auth.OidcClient;
+import com.streamx.cli.auth.Scopes;
+import com.streamx.cli.framework.AbstractSilentCommand;
+import com.streamx.cli.framework.CommandResult;
+import java.util.Optional;
+import picocli.CommandLine;
+
+@CommandLine.Command(
+ name = "login",
+ header = "Log in to StreamX",
+ description = {
+ "Uses the browser (authorization code + PKCE) when a local browser is available,",
+ "and falls back to the device flow over SSH or with --no-browser."
+ }
+)
+public class LoginCommand extends AbstractSilentCommand {
+
+ @CommandLine.Option(
+ names = "--no-browser",
+ description = "Use the device flow instead of opening a local browser "
+ + "(for SSH sessions and headless machines)"
+ )
+ public boolean noBrowser;
+
+ @Override
+ public CommandResult runCommand() {
+ AuthConfig config = AuthConfig.load();
+ OidcClient client = InteractiveGrant.clientFromConfig(config);
+
+ Credentials credentials = InteractiveGrant.run(
+ client, client.discover(), Scopes.LOGIN, noBrowser);
+
+ Optional previous = CredentialsStore.load();
+ CredentialsStore.save(credentials);
+ previous.ifPresent(LoginCommand::revokePrevious);
+
+ System.out.println(msg.authLoginSuccess());
+ return new CommandResult<>(null);
+ }
+
+ private static void revokePrevious(Credentials previous) {
+ if (previous.refreshToken() == null || previous.issuerUrl() == null) {
+ return;
+ }
+ new OidcClient(
+ previous.issuerUrl(),
+ previous.clientId(),
+ previous.insecure())
+ .revokeQuietly(previous.refreshToken());
+ }
+}
diff --git a/src/main/java/com/streamx/cli/commands/auth/logout/LogoutCommand.java b/src/main/java/com/streamx/cli/commands/auth/logout/LogoutCommand.java
new file mode 100644
index 00000000..41ff38b2
--- /dev/null
+++ b/src/main/java/com/streamx/cli/commands/auth/logout/LogoutCommand.java
@@ -0,0 +1,45 @@
+package com.streamx.cli.commands.auth.logout;
+
+import static com.streamx.cli.i18n.MessageProvider.msg;
+
+import com.streamx.cli.auth.Credentials;
+import com.streamx.cli.auth.CredentialsStore;
+import com.streamx.cli.auth.OidcClient;
+import com.streamx.cli.framework.AbstractSilentCommand;
+import com.streamx.cli.framework.CliException;
+import com.streamx.cli.framework.CommandResult;
+import picocli.CommandLine;
+
+@CommandLine.Command(
+ name = "logout",
+ header = "Log out of StreamX"
+)
+public class LogoutCommand extends AbstractSilentCommand {
+ @Override
+ public CommandResult runCommand() {
+ if (!CredentialsStore.exists()) {
+ System.out.println(msg.authLogoutNotLoggedIn());
+ return new CommandResult<>(null);
+ }
+
+ try {
+ CredentialsStore.load().ifPresent(LogoutCommand::revoke);
+ } catch (CliException expected) {
+ }
+
+ CredentialsStore.delete();
+
+ System.out.println(msg.authLogoutSuccess());
+ return new CommandResult<>(null);
+ }
+
+ private static void revoke(Credentials credentials) {
+ if (credentials.refreshToken() != null && credentials.issuerUrl() != null) {
+ new OidcClient(
+ credentials.issuerUrl(),
+ credentials.clientId(),
+ credentials.insecure())
+ .revokeQuietly(credentials.refreshToken());
+ }
+ }
+}
diff --git a/src/main/java/com/streamx/cli/commands/auth/whoami/WhoamiCommand.java b/src/main/java/com/streamx/cli/commands/auth/whoami/WhoamiCommand.java
new file mode 100644
index 00000000..0d988092
--- /dev/null
+++ b/src/main/java/com/streamx/cli/commands/auth/whoami/WhoamiCommand.java
@@ -0,0 +1,109 @@
+package com.streamx.cli.commands.auth.whoami;
+
+import static com.streamx.cli.i18n.MessageProvider.msg;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.streamx.cli.auth.AccessTokenClaims;
+import com.streamx.cli.auth.Credentials;
+import com.streamx.cli.auth.CredentialsStore;
+import com.streamx.cli.auth.Identity;
+import com.streamx.cli.framework.AbstractCommand;
+import com.streamx.cli.framework.CliException;
+import com.streamx.cli.framework.CommandResult;
+import com.streamx.cli.platform.AccessTokens;
+import com.streamx.cli.platform.PlatformClients;
+import com.streamx.cli.platform.ProfileApi;
+import com.streamx.cli.platform.generated.model.Profile;
+import java.time.Instant;
+import java.time.format.DateTimeFormatter;
+import picocli.CommandLine;
+
+@CommandLine.Command(
+ name = "whoami",
+ header = "Display the currently logged in user"
+)
+public class WhoamiCommand extends AbstractCommand {
+
+ @Override
+ public String getTextOutput(CommandResult result) {
+ Identity identity = result.getData();
+ String expires = identity.expiresAt() == null
+ ? "-"
+ : identity.expiresAt() + (identity.expired() ? " (expired)" : "");
+ return """
+ username = %s
+ name = %s
+ email = %s
+ subject = %s
+ issuer = %s
+ expires = %s
+ auth = %s
+ token id = %s"""
+ .formatted(
+ orDash(identity.username()),
+ orDash(identity.name()),
+ orDash(identity.email()),
+ orDash(identity.subject()),
+ orDash(identity.issuer()),
+ expires,
+ AccessTokens.usingPlatformToken() ? "personal access token" : "login session",
+ orDash(identity.tokenId()));
+ }
+
+ private static String orDash(String value) {
+ return value == null ? "-" : value;
+ }
+
+ private static Identity identityFromPlatform() {
+ Profile profile;
+ try (PlatformClients client = PlatformClients.fromConfig()) {
+ profile = new ProfileApi(client).get();
+ }
+ if (profile == null) {
+ throw new CliException(msg.authTokenIdentityUnavailable());
+ }
+ return new Identity(
+ firstNonBlank(profile.getDisplayName(), profile.getEmail()),
+ firstNonBlank(profile.getDisplayName(),
+ join(profile.getFirstName(), profile.getLastName())),
+ profile.getEmail(),
+ profile.getUserId(),
+ null,
+ null,
+ false,
+ AccessTokens.platformTokenId().orElse(null));
+ }
+
+ private static String join(String first, String last) {
+ return firstNonBlank(((first == null ? "" : first) + " "
+ + (last == null ? "" : last)).trim(), null);
+ }
+
+ private static String firstNonBlank(String value, String fallback) {
+ return value == null || value.isBlank() ? fallback : value;
+ }
+
+ @Override
+ public CommandResult runCommand() {
+ if (AccessTokens.usingPlatformToken()) {
+ // A personal access token is opaque and carries no claims: ask the platform who it acts as.
+ return new CommandResult<>(identityFromPlatform());
+ }
+
+ Credentials credentials = CredentialsStore.load()
+ .orElseThrow(() -> new CliException(msg.platformNotLoggedIn()));
+
+ JsonNode claims = AccessTokenClaims.of(credentials.accessToken());
+ Instant expiresAt = credentials.expiresAt();
+
+ return new CommandResult<>(new Identity(
+ claims.path("preferred_username").asText(null),
+ claims.path("name").asText(null),
+ claims.path("email").asText(null),
+ claims.path("sub").asText(null),
+ claims.path("iss").asText(null),
+ expiresAt == null ? null : DateTimeFormatter.ISO_INSTANT.format(expiresAt),
+ expiresAt != null && Instant.now().isAfter(expiresAt),
+ null));
+ }
+}
diff --git a/src/main/java/com/streamx/cli/commands/info/InfoCommand.java b/src/main/java/com/streamx/cli/commands/info/InfoCommand.java
new file mode 100644
index 00000000..f50047ab
--- /dev/null
+++ b/src/main/java/com/streamx/cli/commands/info/InfoCommand.java
@@ -0,0 +1,586 @@
+package com.streamx.cli.commands.info;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.streamx.cli.auth.AuthConfig;
+import com.streamx.cli.auth.Credentials;
+import com.streamx.cli.auth.CredentialsStore;
+import com.streamx.cli.auth.OidcClient;
+import com.streamx.cli.commands.info.InfoResult.Probe;
+import com.streamx.cli.commands.info.InfoResult.Setting;
+import com.streamx.cli.config.StreamxHome;
+import com.streamx.cli.framework.AbstractCommand;
+import com.streamx.cli.framework.CliException;
+import com.streamx.cli.framework.CommandResult;
+import com.streamx.cli.framework.TextTable;
+import com.streamx.cli.framework.Urls;
+import com.streamx.cli.ingestion.IngestionClientConfig;
+import com.streamx.cli.platform.AccessTokens;
+import com.streamx.cli.platform.PlatformConfig;
+import com.streamx.cli.platform.PlatformContext;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.security.GeneralSecurityException;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.Base64;
+import java.util.List;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLException;
+import org.apache.http.client.config.RequestConfig;
+import org.apache.http.client.methods.CloseableHttpResponse;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.conn.ssl.NoopHostnameVerifier;
+import org.apache.http.conn.ssl.TrustAllStrategy;
+import org.apache.http.impl.client.CloseableHttpClient;
+import org.apache.http.impl.client.HttpClientBuilder;
+import org.apache.http.impl.client.HttpClients;
+import org.apache.http.ssl.SSLContexts;
+import org.apache.http.util.EntityUtils;
+import picocli.CommandLine;
+
+@CommandLine.Command(
+ name = "info",
+ header = "Show CLI, context and connectivity diagnostics",
+ description = "Reports the CLI version, the active context and where it came from, the "
+ + "effective endpoint settings, the stored login, and probes the configured endpoints. "
+ + "Useful as the first thing to share when something does not work."
+)
+public class InfoCommand extends AbstractCommand {
+
+ private static final Duration PROBE_TIMEOUT = Duration.ofSeconds(3);
+ private static final ObjectMapper MAPPER = new ObjectMapper();
+
+ private static final String UP = "UP";
+ private static final String DEGRADED = "DEGRADED";
+ private static final String DOWN = "DOWN";
+ private static final String TLS_ERROR = "TLS ERROR";
+ private static final String AUTH_REJECTED = "AUTH REJECTED";
+
+ @CommandLine.Option(
+ names = "--check",
+ description = "Exit with code 1 unless every configured endpoint probes healthy (UP)"
+ )
+ public boolean check;
+
+ /** Diagnostics must keep working when the selected context is broken or missing. */
+ @Override
+ public boolean needsContext() {
+ return false;
+ }
+
+ @Override
+ public CommandResult runCommand() {
+ List warnings = new ArrayList<>();
+
+ InfoResult.Cli cli = new InfoResult.Cli(
+ cliVersion(),
+ runtime(),
+ StreamxHome.getStreamxHome().toString(),
+ StreamxHome.getStreamxHomeSource());
+
+ String active = null;
+ try {
+ active = StreamxHome.getActiveContext();
+ } catch (CliException e) {
+ warnings.add(e.getMessage());
+ }
+ boolean exists = active != null && StreamxHome.contextExists(active);
+ InfoResult.Context context = new InfoResult.Context(
+ active,
+ StreamxHome.getActiveContextSource(),
+ exists,
+ active == null ? null : StreamxHome.getConfigDirOf(active)
+ .resolve("application.properties").toString(),
+ quiet(PlatformContext::effectiveOrg),
+ quiet(PlatformContext::effectiveOrgSource),
+ quiet(PlatformContext::effectiveProject),
+ quiet(PlatformContext::effectiveProjectSource));
+ if (active != null && !exists) {
+ warnings.add("Context '" + active + "' does not exist yet");
+ }
+
+ Properties fileSettings = exists ? loadSettings(active) : new Properties();
+ List settings = collectSettings(fileSettings, warnings);
+
+ InfoResult.Login login = describeLogin(fileSettings, warnings);
+
+ List connectivity = probeEndpoints(fileSettings);
+
+ InfoResult result =
+ new InfoResult(cli, context, settings, login, connectivity, warnings);
+ CommandResult commandResult = new CommandResult<>(result);
+ if (check && connectivity.stream().anyMatch(p -> !UP.equals(p.status()))) {
+ commandResult.setExitCodeOverride(1);
+ }
+ return commandResult;
+ }
+
+ // ---- report assembly -----------------------------------------------------
+
+ private static List collectSettings(Properties file, List warnings) {
+ List settings = new ArrayList<>();
+ settings.add(describeSetting(file, AuthConfig.STREAMX_AUTH_SERVER_URL, null));
+ settings.add(describeSetting(file, AuthConfig.STREAMX_AUTH_REALM, AuthConfig.DEFAULT_REALM));
+ settings.add(describeSetting(
+ file, AuthConfig.STREAMX_AUTH_CLIENT_ID, AuthConfig.DEFAULT_CLIENT_ID));
+ settings.add(describeSetting(file, AuthConfig.STREAMX_AUTH_INSECURE, "false"));
+ settings.add(describeSetting(file, PlatformConfig.STREAMX_PLATFORM_URL, null));
+ settings.add(describeSetting(file, PlatformConfig.STREAMX_PLATFORM_INSECURE, "false"));
+ settings.add(describeSetting(file, IngestionClientConfig.STREAMX_INGESTION_URL, null));
+ settings.add(describeSetting(file, IngestionClientConfig.STREAMX_INGESTION_INSECURE, "false"));
+
+ if (effective(file, AuthConfig.STREAMX_AUTH_SERVER_URL) == null) {
+ warnings.add("Auth server is not configured - run: streamx context configure");
+ }
+ for (String urlKey : List.of(AuthConfig.STREAMX_AUTH_SERVER_URL,
+ PlatformConfig.STREAMX_PLATFORM_URL, IngestionClientConfig.STREAMX_INGESTION_URL)) {
+ String url = effective(file, urlKey);
+ if (url != null && url.startsWith("http://")) {
+ warnings.add(urlKey + " uses cleartext http - credentials may be exposed");
+ }
+ }
+ return settings;
+ }
+
+ private static Setting describeSetting(Properties file, String key, String defaultValue) {
+ String fileValue = file.getProperty(key);
+ String systemValue = System.getProperty(key);
+ if (systemValue != null && !systemValue.equals(fileValue)) {
+ return new Setting(key, systemValue, "system property override");
+ }
+ if (fileValue != null && !fileValue.isBlank()) {
+ return new Setting(key, fileValue, "context");
+ }
+ if (defaultValue != null) {
+ return new Setting(key, defaultValue, "default");
+ }
+ return new Setting(key, null, "not set");
+ }
+
+ private static String effective(Properties file, String key) {
+ String systemValue = System.getProperty(key);
+ if (systemValue != null && !systemValue.isBlank()) {
+ return systemValue;
+ }
+ String fileValue = file.getProperty(key);
+ return fileValue == null || fileValue.isBlank() ? null : fileValue;
+ }
+
+ private InfoResult.Login describeLogin(Properties file, List warnings) {
+ if (AccessTokens.usingPlatformToken()) {
+ // The env credential outranks any stored session; a token is opaque and never expires.
+ return new InfoResult.Login("authenticated with a personal access token", null, null, null);
+ }
+ Optional credentials;
+ try {
+ credentials = CredentialsStore.load();
+ } catch (RuntimeException e) {
+ warnings.add("Stored login is unreadable: " + e.getMessage());
+ return new InfoResult.Login("unreadable", null, null, null);
+ }
+ if (credentials.isEmpty()) {
+ return new InfoResult.Login("not logged in", null, null, null);
+ }
+ Credentials creds = credentials.get();
+ boolean expired = creds.expiresAt() != null && creds.expiresAt().isBefore(Instant.now());
+ String state = expired ? "expired (a refresh is attempted on use)" : "logged in";
+ String user = usernameFromJwt(creds.accessToken());
+
+ String serverUrl = effective(file, AuthConfig.STREAMX_AUTH_SERVER_URL);
+ String realm = Optional.ofNullable(effective(file, AuthConfig.STREAMX_AUTH_REALM))
+ .orElse(AuthConfig.DEFAULT_REALM);
+ if (serverUrl != null && creds.issuerUrl() != null) {
+ String expected = OidcClient.issuerUrl(serverUrl, realm);
+ if (!expected.equals(creds.issuerUrl())) {
+ warnings.add("Stored login belongs to " + creds.issuerUrl()
+ + " but this context is configured for " + expected
+ + " - run: streamx auth login");
+ }
+ }
+ return new InfoResult.Login(
+ state,
+ user,
+ creds.expiresAt() == null ? null : creds.expiresAt().toString(),
+ creds.issuerUrl());
+ }
+
+ /** Display-only decode of the JWT payload; the token is never verified here. */
+ private static String usernameFromJwt(String accessToken) {
+ if (accessToken == null) {
+ return null;
+ }
+ String[] parts = accessToken.split("\\.");
+ if (parts.length < 2) {
+ return null;
+ }
+ try {
+ JsonNode payload =
+ MAPPER.readTree(Base64.getUrlDecoder().decode(parts[1]));
+ String username = payload.path("preferred_username").asText(null);
+ return username != null ? username : payload.path("email").asText(null);
+ } catch (IOException | IllegalArgumentException e) {
+ return null;
+ }
+ }
+
+ // ---- endpoint probes -----------------------------------------------------
+
+ private List probeEndpoints(Properties file) {
+ String authUrl = effective(file, AuthConfig.STREAMX_AUTH_SERVER_URL);
+ String realm = Optional.ofNullable(effective(file, AuthConfig.STREAMX_AUTH_REALM))
+ .orElse(AuthConfig.DEFAULT_REALM);
+ boolean authInsecure = Boolean.parseBoolean(effective(file, AuthConfig.STREAMX_AUTH_INSECURE));
+ String platformUrl = effective(file, PlatformConfig.STREAMX_PLATFORM_URL);
+ boolean platformInsecure =
+ Boolean.parseBoolean(effective(file, PlatformConfig.STREAMX_PLATFORM_INSECURE));
+ Optional credentials;
+ try {
+ credentials = CredentialsStore.load();
+ } catch (RuntimeException unreadable) {
+ credentials = Optional.empty();
+ }
+ String ingestionUrl = effective(file, IngestionClientConfig.STREAMX_INGESTION_URL);
+ boolean ingestionInsecure =
+ Boolean.parseBoolean(effective(file, IngestionClientConfig.STREAMX_INGESTION_INSECURE));
+
+ List> probes = new ArrayList<>();
+ if (authUrl != null) {
+ String discovery = OidcClient.issuerUrl(authUrl, realm) + "/.well-known/openid-configuration";
+ probes.add(() -> probeAuth(discovery, authInsecure));
+ }
+ if (platformUrl != null) {
+ String base = platformUrl.replaceAll("/+$", "");
+ probes.add(() -> probePlatform(base, platformInsecure));
+ }
+ if (ingestionUrl != null) {
+ String base = ingestionUrl.replaceAll("/+$", "");
+ probes.add(() -> probeHealth("ingestion", base, ingestionInsecure));
+ }
+ if (platformUrl != null && (credentials.isPresent() || AccessTokens.usingPlatformToken())) {
+ String base = platformUrl.replaceAll("/+$", "");
+ if (Urls.isCleartextRemote(base)) {
+ probes.add(() -> new Probe("api (authenticated)", base, DEGRADED, null,
+ "not probed: refusing to send a credential over cleartext http"));
+ } else {
+ // AccessTokens prefers the env personal access token, then the stored session.
+ String token = quiet(AccessTokens::current);
+ if (token == null) {
+ probes.add(() -> new Probe("api (authenticated)", base, DEGRADED, null,
+ "not probed: no usable credential - run: streamx auth login"));
+ } else {
+ probes.add(() -> probeAuthenticatedApi(base, platformInsecure, token));
+ }
+ }
+ }
+ if (probes.isEmpty()) {
+ return List.of();
+ }
+
+ ExecutorService pool = Executors.newFixedThreadPool(probes.size());
+ try {
+ List> futures = pool.invokeAll(
+ probes, PROBE_TIMEOUT.toSeconds() + 3, TimeUnit.SECONDS);
+ List results = new ArrayList<>();
+ for (Future future : futures) {
+ try {
+ results.add(future.get());
+ } catch (Exception e) {
+ results.add(new Probe("unknown", null, DOWN, null, "probe did not finish"));
+ }
+ }
+ return results;
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return List.of();
+ } finally {
+ pool.shutdownNow();
+ }
+ }
+
+ private static Probe probeAuth(String discoveryUrl, boolean insecure) {
+ HttpProbe probe = get(discoveryUrl, insecure, null);
+ if (probe.status == 200) {
+ return new Probe("auth", discoveryUrl, UP, probe.latencyMs, "discovery document served");
+ }
+ if (probe.status == 404) {
+ return new Probe("auth", discoveryUrl, DOWN, probe.latencyMs,
+ "server responds but the realm does not exist");
+ }
+ return failure("auth", discoveryUrl, probe);
+ }
+
+ private Probe probePlatform(String baseUrl, boolean insecure) {
+ Probe health = probeHealth("platform", baseUrl, insecure);
+ if (UP.equals(health.status()) || TLS_ERROR.equals(health.status())) {
+ return health;
+ }
+ String apiUrl = baseUrl + "/api/v1/organizations";
+ HttpProbe fallback = get(apiUrl, insecure, null);
+ boolean alive = fallback.status >= 200 && fallback.status < 400
+ || fallback.status == 401 || fallback.status == 403;
+ if (alive) {
+ return new Probe("platform", apiUrl, UP, fallback.latencyMs,
+ "health endpoint unavailable; API answered HTTP " + fallback.status);
+ }
+ if (fallback.status == 404) {
+ return new Probe("platform", apiUrl, DOWN, fallback.latencyMs,
+ "answers HTTP 404 - whatever is at this URL is not the StreamX platform API; "
+ + "check streamx.platform.url");
+ }
+ if (fallback.status > 0) {
+ return new Probe("platform", apiUrl, DOWN, fallback.latencyMs,
+ "HTTP " + fallback.status + " - the endpoint is reachable but the platform "
+ + "is not serving");
+ }
+ return health;
+ }
+
+ private static Probe probeHealth(String name, String baseUrl, boolean insecure) {
+ String url = baseUrl + "/q/health/ready";
+ HttpProbe probe = get(url, insecure, null);
+ if (probe.status == 200) {
+ boolean up = probe.body != null && probe.body.contains("\"UP\"");
+ return new Probe(name, url, up ? UP : DEGRADED, probe.latencyMs,
+ up ? "ready" : "health endpoint reports not ready");
+ }
+ if (probe.status > 0) {
+ return new Probe(name, url, DEGRADED, probe.latencyMs, "HTTP " + probe.status);
+ }
+ return failure(name, url, probe);
+ }
+
+ private static Probe probeAuthenticatedApi(String baseUrl, boolean insecure, String token) {
+ String url = baseUrl + "/api/v1/organizations";
+ HttpProbe probe = get(url, insecure, token);
+ if (probe.status == 200) {
+ String detail = "token accepted";
+ try {
+ JsonNode node = MAPPER.readTree(probe.body);
+ if (node.isArray()) {
+ detail = "token accepted; member of " + node.size() + " organization(s)";
+ }
+ } catch (IOException expected) {
+ }
+ return new Probe("api (authenticated)", url, UP, probe.latencyMs, detail);
+ }
+ if (probe.status == 401 || probe.status == 403) {
+ // 'auth login' is impossible in CI; point at the credential actually in use.
+ String remedy = AccessTokens.usingPlatformToken()
+ ? "check the personal access token in " + AccessTokens.STREAMX_PLATFORM_TOKEN
+ : "run: streamx auth login";
+ return new Probe("api (authenticated)", url, AUTH_REJECTED, probe.latencyMs,
+ "token rejected (HTTP " + probe.status + ") - " + remedy);
+ }
+ if (probe.status > 0) {
+ return new Probe("api (authenticated)", url, DEGRADED, probe.latencyMs,
+ "HTTP " + probe.status);
+ }
+ return failure("api (authenticated)", url, probe);
+ }
+
+ /**
+ * Failed probes never reached the server, so the elapsed time is reported as time-to-failure
+ * in the detail (fast = DNS/refused, ~timeout = blackholed) - not in the latency column.
+ */
+ private static Probe failure(String name, String url, HttpProbe probe) {
+ String after = "failed after " + probe.latencyMs + "ms: ";
+ if (probe.tlsError) {
+ return new Probe(name, url, TLS_ERROR, null,
+ after + probe.error + " - self-signed certs need the matching *.insecure=true setting");
+ }
+ return new Probe(name, url, DOWN, null, after + probe.error);
+ }
+
+ private record HttpProbe(int status, String body, Long latencyMs, String error,
+ boolean tlsError) {
+ }
+
+ private static HttpProbe get(String url, boolean insecure, String bearerToken) {
+ long start = System.nanoTime();
+ try (CloseableHttpClient client = buildHttpClient(insecure, bearerToken == null)) {
+ HttpGet request = new HttpGet(url);
+ request.setHeader("Accept", "application/json");
+ if (bearerToken != null) {
+ request.setHeader("Authorization", "Bearer " + bearerToken);
+ }
+ try (CloseableHttpResponse response = client.execute(request)) {
+ long latency = (System.nanoTime() - start) / 1_000_000;
+ String body = response.getEntity() == null ? null
+ : EntityUtils.toString(response.getEntity(), StandardCharsets.UTF_8);
+ return new HttpProbe(
+ response.getStatusLine().getStatusCode(), body, latency, null, false);
+ }
+ } catch (SSLException e) {
+ return new HttpProbe(0, null, elapsedMs(start), rootMessage(e), true);
+ } catch (IOException | RuntimeException e) {
+ return new HttpProbe(0, null, elapsedMs(start), rootMessage(e), false);
+ }
+ }
+
+ private static long elapsedMs(long startNanos) {
+ return (System.nanoTime() - startNanos) / 1_000_000;
+ }
+
+ private static String rootMessage(Throwable throwable) {
+ Throwable root = throwable;
+ while (root.getCause() != null && root.getCause() != root) {
+ root = root.getCause();
+ }
+ String message = root.getMessage();
+ return message == null || message.isBlank() ? root.getClass().getSimpleName() : message;
+ }
+
+ private static CloseableHttpClient buildHttpClient(boolean insecure, boolean followRedirects) {
+ int timeoutMillis = (int) PROBE_TIMEOUT.toMillis();
+ RequestConfig requestConfig = RequestConfig.custom()
+ .setConnectTimeout(timeoutMillis)
+ .setConnectionRequestTimeout(timeoutMillis)
+ .setSocketTimeout(timeoutMillis)
+ .build();
+ HttpClientBuilder builder = HttpClients.custom()
+ .setDefaultRequestConfig(requestConfig)
+ .disableAutomaticRetries();
+ if (!followRedirects) {
+ builder.disableRedirectHandling();
+ }
+ if (!insecure) {
+ return builder.build();
+ }
+ try {
+ SSLContext sslContext = SSLContexts.custom()
+ .loadTrustMaterial(null, TrustAllStrategy.INSTANCE)
+ .build();
+ return builder
+ .setSSLContext(sslContext)
+ .setSSLHostnameVerifier(NoopHostnameVerifier.INSTANCE)
+ .build();
+ } catch (GeneralSecurityException e) {
+ throw new CliException(e.getMessage(), e);
+ }
+ }
+
+ // ---- helpers -------------------------------------------------------------
+
+ /** "native (Substrate VM, JDK x)" in a native image, "JVM ( )" otherwise. */
+ private static String runtime() {
+ String vm = System.getProperty("java.vm.name", "unknown");
+ String version = System.getProperty("java.version", "");
+ if ("runtime".equals(System.getProperty("org.graalvm.nativeimage.imagecode"))) {
+ return "native (" + vm + ", JDK " + version + ")";
+ }
+ return "JVM (" + vm + " " + version + ")";
+ }
+
+ private static String quiet(java.util.function.Supplier supplier) {
+ try {
+ return supplier.get();
+ } catch (RuntimeException brokenContext) {
+ return null;
+ }
+ }
+
+ private String cliVersion() {
+ try {
+ String[] version = new com.streamx.cli.util.VersionProvider().getVersion();
+ return version.length > 0 ? version[0] : "unknown";
+ } catch (Exception e) {
+ return "unknown";
+ }
+ }
+
+ private static Properties loadSettings(String context) {
+ Properties properties = new Properties();
+ Path path = StreamxHome.getConfigDirOf(context).resolve("application.properties");
+ if (!Files.isRegularFile(path)) {
+ return properties;
+ }
+ try (InputStream in = Files.newInputStream(path)) {
+ properties.load(in);
+ } catch (IOException unreadable) {
+ return properties;
+ }
+ return properties;
+ }
+
+ // ---- text rendering ------------------------------------------------------
+
+ @Override
+ public String getTextOutput(CommandResult result) {
+ InfoResult info = result.getData();
+ StringBuilder sb = new StringBuilder();
+
+ sb.append("CLI\n");
+ row(sb, "version", info.cli().version());
+ row(sb, "runtime", info.cli().runtime());
+ row(sb, "streamx home", info.cli().home() + " (" + info.cli().homeSource() + ")");
+
+ sb.append("\nContext\n");
+ row(sb, "active", valueOrDash(info.context().active())
+ + " (" + info.context().source() + ")");
+ row(sb, "exists", info.context().exists() ? "yes" : "no");
+ row(sb, "settings file", valueOrDash(info.context().settingsFile()));
+ row(sb, "current org", valueOrDash(info.context().currentOrg())
+ + (info.context().currentOrgSource() == null
+ ? "" : " (" + info.context().currentOrgSource() + ")"));
+ row(sb, "current project", valueOrDash(info.context().currentProject())
+ + (info.context().currentProjectSource() == null
+ ? "" : " (" + info.context().currentProjectSource() + ")"));
+
+ sb.append("\nSettings\n");
+ sb.append(TextTable.render(
+ List.of("KEY", "VALUE", "SOURCE"),
+ info.settings().stream()
+ .map(s -> List.of(s.key(), valueOrDash(s.value()), s.source()))
+ .toList()).indent(2));
+
+ sb.append("\nLogin\n");
+ row(sb, "state", info.login().state());
+ row(sb, "user", valueOrDash(info.login().user()));
+ row(sb, "expires", valueOrDash(info.login().expiresAt()));
+ row(sb, "issuer", valueOrDash(info.login().issuer()));
+
+ sb.append("\nConnectivity\n");
+ if (info.connectivity().isEmpty()) {
+ sb.append(" (no endpoints configured)\n");
+ } else {
+ sb.append(TextTable.render(
+ List.of("ENDPOINT", "STATUS", "LATENCY", "DETAIL"),
+ info.connectivity().stream()
+ .map(p -> List.of(
+ p.name(),
+ p.status(),
+ p.latencyMs() == null ? "-" : p.latencyMs() + "ms",
+ valueOrDash(p.detail())))
+ .toList()).indent(2));
+ }
+
+ if (!info.warnings().isEmpty()) {
+ sb.append("\nWarnings\n");
+ for (String warning : info.warnings()) {
+ sb.append(" ! ").append(warning).append('\n');
+ }
+ }
+ return sb.toString().stripTrailing();
+ }
+
+ private static void row(StringBuilder sb, String key, String value) {
+ sb.append(" ").append(String.format("%-16s", key)).append(' ')
+ .append(value == null ? "-" : value).append('\n');
+ }
+
+ private static String valueOrDash(String value) {
+ return value == null || value.isBlank() ? "-" : value;
+ }
+}
diff --git a/src/main/java/com/streamx/cli/commands/info/InfoResult.java b/src/main/java/com/streamx/cli/commands/info/InfoResult.java
new file mode 100644
index 00000000..be3a59eb
--- /dev/null
+++ b/src/main/java/com/streamx/cli/commands/info/InfoResult.java
@@ -0,0 +1,43 @@
+package com.streamx.cli.commands.info;
+
+import io.quarkus.runtime.annotations.RegisterForReflection;
+import java.util.List;
+
+@RegisterForReflection
+public record InfoResult(
+ Cli cli,
+ Context context,
+ List settings,
+ Login login,
+ List connectivity,
+ List warnings
+) {
+
+ @RegisterForReflection
+ public record Cli(String version, String runtime, String home, String homeSource) {
+ }
+
+ @RegisterForReflection
+ public record Context(
+ String active,
+ String source,
+ boolean exists,
+ String settingsFile,
+ String currentOrg,
+ String currentOrgSource,
+ String currentProject,
+ String currentProjectSource) {
+ }
+
+ @RegisterForReflection
+ public record Setting(String key, String value, String source) {
+ }
+
+ @RegisterForReflection
+ public record Login(String state, String user, String expiresAt, String issuer) {
+ }
+
+ @RegisterForReflection
+ public record Probe(String name, String target, String status, Long latencyMs, String detail) {
+ }
+}
diff --git a/src/main/java/com/streamx/cli/commands/settings/SettingsSetKeyCompletionCandidates.java b/src/main/java/com/streamx/cli/commands/settings/SettingsSetKeyCompletionCandidates.java
index de52a140..ac2617f5 100644
--- a/src/main/java/com/streamx/cli/commands/settings/SettingsSetKeyCompletionCandidates.java
+++ b/src/main/java/com/streamx/cli/commands/settings/SettingsSetKeyCompletionCandidates.java
@@ -1,7 +1,9 @@
package com.streamx.cli.commands.settings;
+import com.streamx.cli.auth.AuthConfig;
import com.streamx.cli.commands.publish.event.EventTemplateLoader;
import com.streamx.cli.ingestion.IngestionClientConfig;
+import com.streamx.cli.platform.PlatformConfig;
import com.streamx.runner.config.StreamxBaseConfig;
import java.util.Iterator;
import java.util.List;
@@ -14,6 +16,12 @@ public class SettingsSetKeyCompletionCandidates implements Iterable {
IngestionClientConfig.STREAMX_INGESTION_URL,
IngestionClientConfig.STREAMX_INGESTION_AUTH_TOKEN,
IngestionClientConfig.STREAMX_INGESTION_INSECURE,
+ AuthConfig.STREAMX_AUTH_SERVER_URL,
+ AuthConfig.STREAMX_AUTH_REALM,
+ AuthConfig.STREAMX_AUTH_CLIENT_ID,
+ AuthConfig.STREAMX_AUTH_INSECURE,
+ PlatformConfig.STREAMX_PLATFORM_URL,
+ PlatformConfig.STREAMX_PLATFORM_INSECURE,
StreamxBaseConfig.PN_OBSERVABILITY_ENABLED,
StreamxBaseConfig.PN_OBSERVABILITY_WAIT_FOR_STARTUP,
StreamxBaseConfig.PN_CONTAINER_STARTUP_TIMEOUT_SECONDS,
diff --git a/src/main/java/com/streamx/cli/platform/AccessTokens.java b/src/main/java/com/streamx/cli/platform/AccessTokens.java
new file mode 100644
index 00000000..2e2a32dd
--- /dev/null
+++ b/src/main/java/com/streamx/cli/platform/AccessTokens.java
@@ -0,0 +1,97 @@
+package com.streamx.cli.platform;
+
+import static com.streamx.cli.i18n.MessageProvider.msg;
+
+import com.streamx.cli.auth.Credentials;
+import com.streamx.cli.auth.CredentialsStore;
+import com.streamx.cli.auth.OidcClient;
+import com.streamx.cli.framework.CliException;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.Optional;
+
+public final class AccessTokens {
+ private static final Duration EXPIRY_SKEW = Duration.ofSeconds(30);
+
+ public static final String STREAMX_PLATFORM_TOKEN = "STREAMX_PLATFORM_TOKEN";
+
+ private AccessTokens() {
+ }
+
+ public static String current() {
+ String envToken = platformTokenOverride();
+ if (envToken != null) {
+ return envToken;
+ }
+ Credentials credentials = stored();
+ if (!isExpiring(credentials)) {
+ return credentials.accessToken();
+ }
+ return refresh(credentials);
+ }
+
+ public static String forceRefresh() {
+ String envToken = platformTokenOverride();
+ if (envToken != null) {
+ // A personal access token cannot be refreshed; a 401 means it is invalid or revoked.
+ return envToken;
+ }
+ return refresh(stored());
+ }
+
+ public static boolean usingPlatformToken() {
+ return platformTokenOverride() != null;
+ }
+
+ public static void requireInteractiveSession() {
+ if (usingPlatformToken()) {
+ throw new CliException(msg.authTokenNeedsLoginSession(STREAMX_PLATFORM_TOKEN));
+ }
+ }
+
+ public static Optional platformTokenId() {
+ String token = platformTokenOverride();
+ if (token == null) {
+ return Optional.empty();
+ }
+ String[] parts = token.split("_");
+ return parts.length == 5 && "sxp".equals(parts[0]) && !parts[2].isBlank()
+ ? Optional.of(parts[2])
+ : Optional.empty();
+ }
+
+ private static String platformTokenOverride() {
+ String value = System.getenv(STREAMX_PLATFORM_TOKEN);
+ if (value == null || value.isBlank()) {
+ value = System.getProperty(STREAMX_PLATFORM_TOKEN);
+ }
+ return value == null || value.isBlank() ? null : value.trim();
+ }
+
+ private static Credentials stored() {
+ return CredentialsStore.load()
+ .orElseThrow(() -> new CliException(msg.platformNotLoggedIn()));
+ }
+
+ private static String refresh(Credentials credentials) {
+ if (credentials.refreshToken() == null || credentials.issuerUrl() == null) {
+ throw new CliException(msg.authSessionExpired());
+ }
+
+ // Refresh under the TLS policy the session was created with, not the current config.
+ Credentials refreshed = new OidcClient(
+ credentials.issuerUrl(),
+ credentials.clientId(),
+ credentials.insecure())
+ .refresh(credentials.refreshToken());
+
+ CredentialsStore.save(refreshed);
+ return refreshed.accessToken();
+ }
+
+ private static boolean isExpiring(Credentials credentials) {
+ return Optional.ofNullable(credentials.expiresAt())
+ .map(expiry -> Instant.now().plus(EXPIRY_SKEW).isAfter(expiry))
+ .orElse(true);
+ }
+}
diff --git a/src/main/java/com/streamx/cli/platform/AuthHeaderFilter.java b/src/main/java/com/streamx/cli/platform/AuthHeaderFilter.java
new file mode 100644
index 00000000..efc67b0a
--- /dev/null
+++ b/src/main/java/com/streamx/cli/platform/AuthHeaderFilter.java
@@ -0,0 +1,14 @@
+package com.streamx.cli.platform;
+
+import jakarta.ws.rs.client.ClientRequestContext;
+import jakarta.ws.rs.client.ClientRequestFilter;
+import jakarta.ws.rs.core.HttpHeaders;
+
+public class AuthHeaderFilter implements ClientRequestFilter {
+
+ @Override
+ public void filter(ClientRequestContext context) {
+ context.getHeaders().putSingle(HttpHeaders.AUTHORIZATION, "Bearer " + AccessTokens.current());
+ context.getHeaders().putSingle(HttpHeaders.ACCEPT, "application/json");
+ }
+}
diff --git a/src/main/java/com/streamx/cli/platform/PlatformClients.java b/src/main/java/com/streamx/cli/platform/PlatformClients.java
new file mode 100644
index 00000000..7de03229
--- /dev/null
+++ b/src/main/java/com/streamx/cli/platform/PlatformClients.java
@@ -0,0 +1,191 @@
+package com.streamx.cli.platform;
+
+import static com.streamx.cli.i18n.MessageProvider.msg;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializationFeature;
+import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
+import com.streamx.cli.framework.CliException;
+import com.streamx.cli.framework.Urls;
+import io.quarkus.rest.client.reactive.QuarkusRestClientBuilder;
+import jakarta.ws.rs.WebApplicationException;
+import jakarta.ws.rs.core.Response;
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
+
+public class PlatformClients implements AutoCloseable {
+
+ private static final ObjectMapper MAPPER = new ObjectMapper()
+ .registerModule(new JavaTimeModule())
+ .disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS);
+ private static final long TIMEOUT_MS = 30_000;
+ private static final long COMPLETION_TIMEOUT_MS = 3_000;
+
+ private final URI baseUri;
+ private final boolean insecure;
+ private final long timeoutMs;
+ private final List built = new ArrayList<>();
+
+ PlatformClients(String url, boolean insecure, long timeoutMs) {
+ if (Urls.isCleartextRemote(url)) {
+ throw new CliException(msg.platformCleartextHttpBlocked(url));
+ }
+ this.baseUri = URI.create(url.endsWith("/") ? url.substring(0, url.length() - 1) : url);
+ this.insecure = insecure;
+ this.timeoutMs = timeoutMs;
+ }
+
+ public static PlatformClients fromConfig() {
+ return create(TIMEOUT_MS);
+ }
+
+ public static PlatformClients completion() {
+ return create(COMPLETION_TIMEOUT_MS);
+ }
+
+ private static PlatformClients create(long timeoutMs) {
+ PlatformConfig config = PlatformConfig.load();
+ String url = config.url()
+ .filter(value -> !value.isBlank())
+ .orElseThrow(() -> new CliException(
+ msg.platformUrlNotConfigured(PlatformConfig.STREAMX_PLATFORM_URL)));
+ return new PlatformClients(url, config.insecure(), timeoutMs);
+ }
+
+ public T api(Class apiType) {
+ QuarkusRestClientBuilder builder = QuarkusRestClientBuilder.newBuilder()
+ .baseUri(baseUri)
+ .connectTimeout(timeoutMs, TimeUnit.MILLISECONDS)
+ .readTimeout(timeoutMs, TimeUnit.MILLISECONDS)
+ .register(AuthHeaderFilter.class);
+ if (insecure) {
+ builder.trustAll(true).verifyHost(false);
+ }
+ T client = builder.build(apiType);
+ if (client instanceof AutoCloseable closeable) {
+ built.add(closeable);
+ }
+ return client;
+ }
+
+ @Override
+ public void close() {
+ for (AutoCloseable client : built) {
+ try {
+ client.close();
+ } catch (Exception ignored) {
+ continue;
+ }
+ }
+ built.clear();
+ }
+
+ public T call(Supplier operation, Class type) {
+ Response response = invoke(operation);
+ // A personal access token cannot be refreshed, so retrying would just re-send the same
+ // rejected credential; only a login session is worth a second attempt.
+ if (response.getStatus() == 401 && !AccessTokens.usingPlatformToken()) {
+ AccessTokens.forceRefresh();
+ response = invoke(operation);
+ }
+ return handle(response, type);
+ }
+
+ public void call(Supplier operation) {
+ call(operation, null);
+ }
+
+ public List callList(Supplier operation, Class type) {
+ JsonNode array = call(operation, JsonNode.class);
+ List items = new ArrayList<>();
+ if (array != null && array.isArray()) {
+ for (JsonNode node : array) {
+ items.add(MAPPER.convertValue(node, type));
+ }
+ }
+ return items;
+ }
+
+ private Response invoke(Supplier operation) {
+ try {
+ return operation.get();
+ } catch (WebApplicationException errorStatus) {
+ return errorStatus.getResponse();
+ } catch (RuntimeException failure) {
+ throw asCliException(failure);
+ }
+ }
+
+ private T handle(Response response, Class type) {
+ int status = response.getStatus();
+ if (status == 404) {
+ throw new NotFoundException(msg.platformNotFound());
+ }
+ if (status < 200 || status >= 300) {
+ String body = response.hasEntity() ? response.readEntity(String.class) : "";
+ throw new CliException(errorMessage(status, body));
+ }
+ if (type == null || !response.hasEntity()) {
+ return null;
+ }
+ return response.readEntity(type);
+ }
+
+ private CliException asCliException(RuntimeException failure) {
+ for (Throwable cause = failure; cause != null; cause = cause.getCause()) {
+ if (cause instanceof CliException cli) {
+ return cli;
+ }
+ }
+ return new CliException(
+ msg.platformRequestFailed(baseUri.toString(), failure.getMessage()), failure);
+ }
+
+ private String errorMessage(int status, String body) {
+ String detail = extractErrorMessage(body);
+ return switch (status) {
+ // Keep the server's explanation (e.g. "personal access tokens cannot manage tokens"),
+ // and point at the credential actually in use rather than always at 'auth login'.
+ case 401 -> withDetail(AccessTokens.usingPlatformToken()
+ ? msg.platformTokenUnauthorized() : msg.platformUnauthorized(), detail);
+ case 403 -> withDetail(msg.platformAccessDenied(), detail);
+ default -> detail == null
+ ? msg.platformRequestFailedWithStatus(baseUri.toString(), status)
+ : msg.platformRequestRejected(status, detail);
+ };
+ }
+
+ private static String withDetail(String message, String detail) {
+ return detail == null || detail.isBlank() ? message : message + "\n " + detail;
+ }
+
+ private static String extractErrorMessage(String body) {
+ if (body == null || body.isBlank()) {
+ return null;
+ }
+ try {
+ JsonNode node = MAPPER.readTree(body);
+ String message = node.path("errorMessage").asText(null);
+ StringBuilder detail = new StringBuilder(message == null ? "" : message);
+ for (JsonNode violation : node.path("violations")) {
+ detail.append("\n ")
+ .append(violation.path("field").asText(""))
+ .append(": ")
+ .append(violation.path("message").asText(""));
+ }
+ return detail.isEmpty() ? null : detail.toString();
+ } catch (com.fasterxml.jackson.core.JsonProcessingException notJson) {
+ return null;
+ }
+ }
+
+ public static class NotFoundException extends CliException {
+ public NotFoundException(String message) {
+ super(message);
+ }
+ }
+}
diff --git a/src/main/java/com/streamx/cli/platform/ProfileApi.java b/src/main/java/com/streamx/cli/platform/ProfileApi.java
new file mode 100644
index 00000000..ffa83c63
--- /dev/null
+++ b/src/main/java/com/streamx/cli/platform/ProfileApi.java
@@ -0,0 +1,19 @@
+package com.streamx.cli.platform;
+
+import com.streamx.cli.platform.generated.api.ProfileResourceApi;
+import com.streamx.cli.platform.generated.model.Profile;
+
+public class ProfileApi {
+
+ private final PlatformClients clients;
+ private final ProfileResourceApi api;
+
+ public ProfileApi(PlatformClients clients) {
+ this.clients = clients;
+ this.api = clients.api(ProfileResourceApi.class);
+ }
+
+ public Profile get() {
+ return clients.call(() -> api.getProfile(null, null), Profile.class);
+ }
+}
diff --git a/src/main/resources/application.properties b/src/main/resources/application.properties
index 3a541d51..d21c0d53 100644
--- a/src/main/resources/application.properties
+++ b/src/main/resources/application.properties
@@ -32,3 +32,11 @@ quarkus.native.additional-build-args=\
# Fixes "Cannot load required properties from maven-build.properties" error
# when run streamx-runner with native-image executable.
quarkus.native.resources.includes=maven-build.properties,default-event-templates/**
+# OpenAPI-generated platform client (spec unpacked from streamx-platform-aggregated-openapi)
+quarkus.openapi-generator.codegen.spec.openapi_yaml.base-package=com.streamx.cli.platform.generated
+# The aggregated spec repeats the X-Permission-Check header on every operation (invalid per the
+# OpenAPI param-uniqueness rule); skip validation so codegen proceeds and dedupes it.
+quarkus.openapi-generator.codegen.validateSpec=false
+# Every operation returns jakarta.ws.rs.core.Response so the client factory has one uniform
+# path for auth-retry, error mapping and body deserialization.
+quarkus.openapi-generator.codegen.spec.openapi_yaml.return-response=true
diff --git a/src/test/java/com/streamx/cli/auth/OidcClientCleartextTest.java b/src/test/java/com/streamx/cli/auth/OidcClientCleartextTest.java
new file mode 100644
index 00000000..1d17b6de
--- /dev/null
+++ b/src/test/java/com/streamx/cli/auth/OidcClientCleartextTest.java
@@ -0,0 +1,22 @@
+package com.streamx.cli.auth;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import com.streamx.cli.framework.CliException;
+import org.junit.jupiter.api.Test;
+
+class OidcClientCleartextTest {
+
+ @Test
+ void refusesCleartextRemoteAuthServer() {
+ assertThatThrownBy(() -> new OidcClient("http://idp.example.com/realms/x", "cli", false))
+ .isInstanceOf(CliException.class)
+ .hasMessageContaining("cleartext HTTP");
+ }
+
+ @Test
+ void allowsLoopbackHttpForLocalDevelopment() {
+ assertThat(new OidcClient("http://127.0.0.1:8080/realms/x", "cli", false)).isNotNull();
+ }
+}
diff --git a/src/test/java/com/streamx/cli/commands/auth/AuthCodeFlowIT.java b/src/test/java/com/streamx/cli/commands/auth/AuthCodeFlowIT.java
new file mode 100644
index 00000000..6785e399
--- /dev/null
+++ b/src/test/java/com/streamx/cli/commands/auth/AuthCodeFlowIT.java
@@ -0,0 +1,109 @@
+package com.streamx.cli.commands.auth;
+
+import static com.streamx.cli.i18n.MessageProvider.msg;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import com.streamx.cli.auth.OidcAuthCodeFlow;
+import com.streamx.cli.auth.OidcClient;
+import com.streamx.cli.auth.OidcClient.Endpoints;
+import java.io.IOException;
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+class AuthCodeFlowIT {
+
+ private static final String REALM = "streamx";
+
+ private StubOidcServer oidcServer;
+ private Endpoints endpoints;
+
+ @BeforeEach
+ void setUp() throws IOException {
+ oidcServer = new StubOidcServer(REALM, 0);
+ endpoints = new OidcClient(
+ oidcServer.getServerUrl() + "/realms/" + REALM, "streamx-cli", false).discover();
+ }
+
+ @AfterEach
+ void tearDown() {
+ if (oidcServer != null) {
+ oidcServer.close();
+ }
+ }
+
+ private static void simulateBrowser(String authorizationUrl) {
+ HttpClient http = HttpClient.newBuilder().followRedirects(HttpClient.Redirect.NEVER).build();
+ try {
+ HttpResponse response = http.send(
+ HttpRequest.newBuilder(URI.create(authorizationUrl)).GET().build(),
+ HttpResponse.BodyHandlers.discarding());
+ String location = response.headers().firstValue("Location").orElseThrow();
+ http.send(HttpRequest.newBuilder(URI.create(location)).GET().build(),
+ HttpResponse.BodyHandlers.discarding());
+ } catch (IOException | InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Test
+ void shouldCompletePkceLoginWithoutACode() {
+ OidcClient client = new OidcClient(
+ oidcServer.getServerUrl() + "/realms/" + REALM, "streamx-cli", false);
+
+ var flow = new OidcAuthCodeFlow(client, AuthCodeFlowIT::simulateBrowser);
+ var credentials = flow.login(endpoints);
+
+ assertThat(credentials.accessToken()).isEqualTo(StubOidcServer.ACCESS_TOKEN);
+ assertThat(credentials.refreshToken()).isEqualTo(StubOidcServer.REFRESH_TOKEN);
+ }
+
+ @Test
+ void shouldSendPkceChallengeAndVerifier() {
+ OidcClient client = new OidcClient(
+ oidcServer.getServerUrl() + "/realms/" + REALM, "streamx-cli", false);
+
+ new OidcAuthCodeFlow(client, AuthCodeFlowIT::simulateBrowser).login(endpoints);
+
+ assertThat(oidcServer.getLastAuthorizationRequest())
+ .containsEntry("code_challenge_method", "S256")
+ .containsKey("code_challenge");
+ assertThat(oidcServer.getLastAuthorizationRequest().get("redirect_uri"))
+ .startsWith("http://127.0.0.1:");
+ assertThat(oidcServer.getLastTokenRequestBody())
+ .contains("grant_type=authorization_code")
+ .contains("code_verifier=")
+ .contains("code=" + StubOidcServer.AUTH_CODE);
+ }
+
+ @Test
+ void shouldFailWhenAuthorizationIsDenied() {
+ oidcServer.denyAuthorization();
+ OidcClient client = new OidcClient(
+ oidcServer.getServerUrl() + "/realms/" + REALM, "streamx-cli", false);
+
+ OidcAuthCodeFlow flow = new OidcAuthCodeFlow(client, AuthCodeFlowIT::simulateBrowser);
+
+ assertThat(org.junit.jupiter.api.Assertions.assertThrows(
+ RuntimeException.class, () -> flow.login(endpoints)).getMessage())
+ .contains(msg.authLoginDenied());
+ }
+
+ @Test
+ void shouldSignalBrowserUnavailableWhenLauncherFails() {
+ OidcClient client = new OidcClient(
+ oidcServer.getServerUrl() + "/realms/" + REALM, "streamx-cli", false);
+
+ OidcAuthCodeFlow.BrowserLauncher failing = url -> {
+ throw new java.io.IOException("no browser");
+ };
+ OidcAuthCodeFlow flow = new OidcAuthCodeFlow(client, failing);
+
+ org.junit.jupiter.api.Assertions.assertThrows(
+ OidcAuthCodeFlow.BrowserUnavailableException.class, () -> flow.login(endpoints));
+ }
+}
diff --git a/src/test/java/com/streamx/cli/commands/auth/AuthCommandIT.java b/src/test/java/com/streamx/cli/commands/auth/AuthCommandIT.java
new file mode 100644
index 00000000..efc17558
--- /dev/null
+++ b/src/test/java/com/streamx/cli/commands/auth/AuthCommandIT.java
@@ -0,0 +1,263 @@
+package com.streamx.cli.commands.auth;
+
+import static com.streamx.cli.i18n.MessageProvider.msg;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import com.streamx.cli.auth.AuthConfig;
+import com.streamx.cli.test.CliBaseIT;
+import io.quarkus.test.junit.QuarkusTest;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.attribute.PosixFilePermissions;
+import java.util.Properties;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+@QuarkusTest
+class AuthCommandIT extends CliBaseIT {
+ private static final String REALM = "streamx";
+
+ private StubOidcServer oidcServer;
+
+ private Path getCredentialsPath() {
+ return streamxHome.resolve("contexts/default/config/credentials.json");
+ }
+
+ private void writeAuthConfig(String serverUrl) throws IOException {
+ Properties properties = new Properties();
+ if (serverUrl != null) {
+ properties.setProperty(AuthConfig.STREAMX_AUTH_SERVER_URL, serverUrl);
+ properties.setProperty(AuthConfig.STREAMX_AUTH_REALM, REALM);
+ properties.setProperty(AuthConfig.STREAMX_AUTH_CLIENT_ID, "streamx-cli");
+ }
+ Path configFile = getConfigPath();
+ Files.createDirectories(configFile.getParent());
+ try (OutputStream out = Files.newOutputStream(configFile)) {
+ properties.store(out, null);
+ }
+ }
+
+ @BeforeEach
+ void cleanState() throws IOException {
+ Files.deleteIfExists(getCredentialsPath());
+ Files.deleteIfExists(getConfigPath());
+ }
+
+ @AfterEach
+ void stopServer() {
+ if (oidcServer != null) {
+ oidcServer.close();
+ oidcServer = null;
+ }
+ }
+
+ @Test
+ void shouldPrintUserCodeAndStoreCredentialsOnLogin() throws Exception {
+ oidcServer = new StubOidcServer(REALM, 0);
+ writeAuthConfig(oidcServer.getServerUrl());
+
+ ProcessResult result = exec("auth", "login", "--no-browser");
+
+ result.assertSuccess();
+ assertThat(result.stderr()).contains(StubOidcServer.USER_CODE);
+ assertThat(result.stdout()).contains(msg.authLoginSuccess());
+
+ assertThat(getCredentialsPath()).exists();
+ String credentials = Files.readString(getCredentialsPath());
+ assertThat(credentials).contains(StubOidcServer.ACCESS_TOKEN);
+ assertThat(credentials).contains(StubOidcServer.REFRESH_TOKEN);
+ }
+
+ @Test
+ void shouldKeepPollingWhileAuthorizationIsPending() throws Exception {
+ oidcServer = new StubOidcServer(REALM, 2);
+ writeAuthConfig(oidcServer.getServerUrl());
+
+ ProcessResult result = exec("auth", "login", "--no-browser");
+
+ result.assertSuccess();
+ assertThat(oidcServer.getTokenRequestCount()).isEqualTo(3);
+ assertThat(getCredentialsPath()).exists();
+ }
+
+ @Test
+ void shouldStoreCredentialsOwnerReadableOnly() throws Exception {
+ oidcServer = new StubOidcServer(REALM, 0);
+ writeAuthConfig(oidcServer.getServerUrl());
+
+ exec("auth", "login", "--no-browser").assertSuccess();
+
+ assertThat(PosixFilePermissions.toString(Files.getPosixFilePermissions(getCredentialsPath())))
+ .isEqualTo("rw-------");
+ }
+
+ @Test
+ void shouldReportLoggedInUserForWhoami() throws Exception {
+ oidcServer = new StubOidcServer(REALM, 0);
+ writeAuthConfig(oidcServer.getServerUrl());
+ exec("auth", "login", "--no-browser").assertSuccess();
+
+ ProcessResult result = exec("auth", "whoami");
+
+ result.assertSuccess();
+ assertThat(result.stdout()).contains("username = " + StubOidcServer.USERNAME);
+ assertThat(result.stdout()).contains("email = " + StubOidcServer.EMAIL);
+ assertThat(result.stdout()).contains("subject = " + StubOidcServer.SUBJECT);
+ assertThat(result.stdout()).contains("expires = ");
+ }
+
+ @Test
+ void shouldReportWhoamiWhenIdentityProviderIsUnreachable() throws Exception {
+ oidcServer = new StubOidcServer(REALM, 0);
+ writeAuthConfig(oidcServer.getServerUrl());
+ exec("auth", "login", "--no-browser").assertSuccess();
+
+ oidcServer.close();
+ oidcServer = null;
+
+ ProcessResult result = exec("auth", "whoami");
+
+ result.assertSuccess();
+ assertThat(result.stdout()).contains("username = " + StubOidcServer.USERNAME);
+ }
+
+ @Test
+ void shouldReportWhoamiAsJson() throws Exception {
+ oidcServer = new StubOidcServer(REALM, 0);
+ writeAuthConfig(oidcServer.getServerUrl());
+ exec("auth", "login", "--no-browser").assertSuccess();
+
+ ProcessResult result = exec("auth", "whoami", "--output", "json");
+
+ result.assertSuccess();
+ assertThat(result.stdout()).contains("\"username\" : \"" + StubOidcServer.USERNAME + "\"");
+ assertThat(result.stdout()).contains("\"subject\" : \"" + StubOidcServer.SUBJECT + "\"");
+ }
+
+ @Test
+ void shouldFailWhoamiWhenNotLoggedIn() throws Exception {
+ ProcessResult result = exec("auth", "whoami");
+
+ result.assertExitCode(1);
+ assertThat(result.stderr()).contains(msg.platformNotLoggedIn());
+ }
+
+ @Test
+ void shouldFailWhenServerUrlNotConfigured() throws Exception {
+ writeAuthConfig(null);
+
+ ProcessResult result = exec("auth", "login", "--no-browser");
+
+ result.assertExitCode(1);
+ assertThat(result.stderr())
+ .contains(msg.authServerUrlNotConfigured(AuthConfig.STREAMX_AUTH_SERVER_URL));
+ assertThat(getCredentialsPath()).doesNotExist();
+ }
+
+ @Test
+ void shouldNotTreatServerErrorAsSuccessfulLogin() throws Exception {
+ oidcServer = new StubOidcServer(REALM, 0);
+ oidcServer.failTokenWithStatus(503, "{}");
+ writeAuthConfig(oidcServer.getServerUrl());
+
+ ProcessResult result = exec("auth", "login", "--no-browser");
+
+ result.assertExitCode(1);
+ assertThat(getCredentialsPath()).doesNotExist();
+ }
+
+ @Test
+ void shouldRejectTokenResponseWithoutAccessToken() throws Exception {
+ oidcServer = new StubOidcServer(REALM, 0);
+ oidcServer.failTokenWithStatus(200, "{\"token_type\":\"Bearer\"}");
+ writeAuthConfig(oidcServer.getServerUrl());
+
+ ProcessResult result = exec("auth", "login", "--no-browser");
+
+ result.assertExitCode(1);
+ assertThat(result.stderr()).contains(msg.authTokenResponseIncomplete());
+ assertThat(getCredentialsPath()).doesNotExist();
+ }
+
+ @Test
+ void shouldRejectMismatchedDiscoveryIssuer() throws Exception {
+ oidcServer = new StubOidcServer(REALM, 0);
+ oidcServer.returnWrongIssuer();
+ writeAuthConfig(oidcServer.getServerUrl());
+
+ ProcessResult result = exec("auth", "login", "--no-browser");
+
+ result.assertExitCode(1);
+ assertThat(result.stderr()).contains("does not match");
+ assertThat(getCredentialsPath()).doesNotExist();
+ }
+
+ @Test
+ void shouldFailWhenLoginIsDenied() throws Exception {
+ oidcServer = new StubOidcServer(REALM, 0);
+ oidcServer.failTokenWith("access_denied");
+ writeAuthConfig(oidcServer.getServerUrl());
+
+ ProcessResult result = exec("auth", "login", "--no-browser");
+
+ result.assertExitCode(1);
+ assertThat(result.stderr()).contains(msg.authLoginDenied());
+ assertThat(getCredentialsPath()).doesNotExist();
+ }
+
+ @Test
+ void shouldRevokeRefreshTokenAndRemoveCredentialsOnLogout() throws Exception {
+ oidcServer = new StubOidcServer(REALM, 0);
+ writeAuthConfig(oidcServer.getServerUrl());
+ exec("auth", "login", "--no-browser").assertSuccess();
+
+ ProcessResult result = exec("auth", "logout");
+
+ result.assertSuccess();
+ assertThat(result.stdout()).contains(msg.authLogoutSuccess());
+ assertThat(getCredentialsPath()).doesNotExist();
+ assertThat(oidcServer.getRevokedTokens()).containsExactly(StubOidcServer.REFRESH_TOKEN);
+ }
+
+ @Test
+ void shouldReportNotLoggedInWhenLoggingOutWithoutCredentials() throws Exception {
+ writeAuthConfig(null);
+
+ ProcessResult result = exec("auth", "logout");
+
+ result.assertSuccess();
+ assertThat(result.stdout()).contains(msg.authLogoutNotLoggedIn());
+ }
+
+ @Test
+ void shouldRemoveCredentialsEvenWhenIdentityProviderIsUnreachable() throws Exception {
+ oidcServer = new StubOidcServer(REALM, 0);
+ writeAuthConfig(oidcServer.getServerUrl());
+ exec("auth", "login", "--no-browser").assertSuccess();
+
+ oidcServer.close();
+ oidcServer = null;
+
+ ProcessResult result = exec("auth", "logout");
+
+ result.assertSuccess();
+ assertThat(getCredentialsPath()).doesNotExist();
+ }
+
+ @Test
+ void shouldClearCorruptCredentialsOnLogout() throws Exception {
+ writeAuthConfig(null);
+ Path credentials = getCredentialsPath();
+ Files.createDirectories(credentials.getParent());
+ Files.writeString(credentials, "{ not valid json");
+
+ ProcessResult result = exec("auth", "logout");
+
+ result.assertSuccess();
+ assertThat(result.stdout()).contains(msg.authLogoutSuccess());
+ assertThat(credentials).doesNotExist();
+ }
+}
diff --git a/src/test/java/com/streamx/cli/commands/auth/AuthCommandRealKeycloakIT.java b/src/test/java/com/streamx/cli/commands/auth/AuthCommandRealKeycloakIT.java
new file mode 100644
index 00000000..7ff728f6
--- /dev/null
+++ b/src/test/java/com/streamx/cli/commands/auth/AuthCommandRealKeycloakIT.java
@@ -0,0 +1,146 @@
+package com.streamx.cli.commands.auth;
+
+import static com.streamx.cli.i18n.MessageProvider.msg;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.streamx.cli.auth.AuthConfig;
+import com.streamx.cli.test.CliBaseIT;
+import io.quarkus.test.junit.QuarkusTest;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Base64;
+import java.util.Properties;
+import java.util.concurrent.TimeUnit;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import org.awaitility.Awaitility;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.EnabledIfSystemProperty;
+
+@QuarkusTest
+@EnabledIfSystemProperty(
+ named = AuthTestEndpoints.SERVER_URL_PROPERTY,
+ matches = ".+",
+ disabledReason = "Set -D" + AuthTestEndpoints.SERVER_URL_PROPERTY + " to run against a real IdP"
+)
+class AuthCommandRealKeycloakIT extends CliBaseIT {
+ private static final Pattern VERIFICATION_LINK =
+ Pattern.compile("(https?://\\S*user_code=[A-Za-z0-9-]+)");
+
+ private static final ObjectMapper MAPPER = new ObjectMapper();
+
+ private Path getCredentialsPath() {
+ return streamxHome.resolve("contexts/default/config/credentials.json");
+ }
+
+ @BeforeEach
+ void configureAgainstRealIdentityProvider() throws IOException {
+ Files.deleteIfExists(getCredentialsPath());
+
+ Properties properties = new Properties();
+ properties.setProperty(AuthConfig.STREAMX_AUTH_SERVER_URL, AuthTestEndpoints.serverUrl());
+ properties.setProperty(AuthConfig.STREAMX_AUTH_REALM, AuthTestEndpoints.realm());
+ properties.setProperty(AuthConfig.STREAMX_AUTH_CLIENT_ID, AuthTestEndpoints.clientId());
+ properties.setProperty(
+ AuthConfig.STREAMX_AUTH_INSECURE, String.valueOf(AuthTestEndpoints.insecure()));
+
+ Path configFile = getConfigPath();
+ Files.createDirectories(configFile.getParent());
+ try (OutputStream out = Files.newOutputStream(configFile)) {
+ properties.store(out, null);
+ }
+ }
+
+ @Test
+ void shouldCompleteDeviceFlowAgainstRealKeycloak() throws Exception {
+ AsyncProcessHandle login = execAsync("auth", "login", "--no-browser");
+
+ String verificationUri = awaitVerificationUri(login);
+ try (KeycloakDeviceApprover approver =
+ new KeycloakDeviceApprover(AuthTestEndpoints.insecure())) {
+ approver.approve(
+ verificationUri, AuthTestEndpoints.username(), AuthTestEndpoints.password());
+ }
+
+ Awaitility.await("cli stores credentials after approval")
+ .atMost(60, TimeUnit.SECONDS)
+ .pollInterval(1, TimeUnit.SECONDS)
+ .until(() -> Files.exists(getCredentialsPath()));
+
+ login.interruptAndJoin(TimeUnit.SECONDS.toMillis(10));
+ assertThat(login.getStdout()).contains(msg.authLoginSuccess());
+
+ JsonNode claims = accessTokenClaims();
+ assertThat(claims.path("iss").asText())
+ .isEqualTo(AuthTestEndpoints.serverUrl() + "/realms/" + AuthTestEndpoints.realm());
+ assertThat(claims.path("azp").asText()).isEqualTo(AuthTestEndpoints.clientId());
+ assertThat(claims.path("preferred_username").asText())
+ .isEqualTo(AuthTestEndpoints.username());
+ }
+
+ @Test
+ void shouldRevokeRealRefreshTokenOnLogout() throws Exception {
+ AsyncProcessHandle login = execAsync("auth", "login", "--no-browser");
+ String verificationUri = awaitVerificationUri(login);
+ try (KeycloakDeviceApprover approver =
+ new KeycloakDeviceApprover(AuthTestEndpoints.insecure())) {
+ approver.approve(
+ verificationUri, AuthTestEndpoints.username(), AuthTestEndpoints.password());
+ }
+ Awaitility.await()
+ .atMost(60, TimeUnit.SECONDS)
+ .pollInterval(1, TimeUnit.SECONDS)
+ .until(() -> Files.exists(getCredentialsPath()));
+ login.interruptAndJoin(TimeUnit.SECONDS.toMillis(10));
+
+ String refreshToken = credentials().path("refresh_token").asText();
+ assertThat(refreshTokenAccepted(refreshToken))
+ .as("refresh token should work before logout")
+ .isTrue();
+
+ exec("auth", "logout").assertSuccess();
+
+ assertThat(getCredentialsPath()).doesNotExist();
+ assertThat(refreshTokenAccepted(refreshToken))
+ .as("refresh token should be revoked at the identity provider after logout")
+ .isFalse();
+ }
+
+ private String awaitVerificationUri(AsyncProcessHandle login) {
+ return Awaitility.await("cli prints the verification link")
+ .atMost(60, TimeUnit.SECONDS)
+ .pollInterval(500, TimeUnit.MILLISECONDS)
+ .until(() -> {
+ Matcher matcher = VERIFICATION_LINK.matcher(login.getStderr() + "\n" + login.getStdout());
+ return matcher.find() ? matcher.group(1) : null;
+ }, uri -> uri != null);
+ }
+
+ private JsonNode credentials() throws IOException {
+ return MAPPER.readTree(Files.readString(getCredentialsPath()));
+ }
+
+ private JsonNode accessTokenClaims() throws IOException {
+ String payload = credentials().path("access_token").asText().split("\\.")[1];
+ return MAPPER.readTree(Base64.getUrlDecoder().decode(payload));
+ }
+
+ private boolean refreshTokenAccepted(String refreshToken) throws Exception {
+ String tokenEndpoint = AuthTestEndpoints.serverUrl()
+ + "/realms/" + AuthTestEndpoints.realm()
+ + "/protocol/openid-connect/token";
+ String form = "grant_type=refresh_token"
+ + "&client_id=" + AuthTestEndpoints.clientId()
+ + "&refresh_token=" + java.net.URLEncoder.encode(refreshToken, StandardCharsets.UTF_8);
+
+ try (var client = new TestHttpClient(AuthTestEndpoints.insecure())) {
+ return client.postForm(tokenEndpoint, form).contains("access_token");
+ }
+ }
+}
diff --git a/src/test/java/com/streamx/cli/commands/auth/AuthTestEndpoints.java b/src/test/java/com/streamx/cli/commands/auth/AuthTestEndpoints.java
new file mode 100644
index 00000000..fd7c4c34
--- /dev/null
+++ b/src/test/java/com/streamx/cli/commands/auth/AuthTestEndpoints.java
@@ -0,0 +1,39 @@
+package com.streamx.cli.commands.auth;
+
+import com.streamx.cli.auth.AuthConfig;
+
+public final class AuthTestEndpoints {
+ public static final String SERVER_URL_PROPERTY = "streamx.test.auth.server-url";
+ public static final String REALM_PROPERTY = "streamx.test.auth.realm";
+ public static final String CLIENT_ID_PROPERTY = "streamx.test.auth.client-id";
+ public static final String USERNAME_PROPERTY = "streamx.test.auth.username";
+ public static final String PASSWORD_PROPERTY = "streamx.test.auth.password";
+ public static final String INSECURE_PROPERTY = "streamx.test.auth.insecure";
+
+ private AuthTestEndpoints() {
+ }
+
+ public static String serverUrl() {
+ return System.getProperty(SERVER_URL_PROPERTY);
+ }
+
+ public static String realm() {
+ return System.getProperty(REALM_PROPERTY, AuthConfig.DEFAULT_REALM);
+ }
+
+ public static String clientId() {
+ return System.getProperty(CLIENT_ID_PROPERTY, AuthConfig.DEFAULT_CLIENT_ID);
+ }
+
+ public static String username() {
+ return System.getProperty(USERNAME_PROPERTY, "user1");
+ }
+
+ public static String password() {
+ return System.getProperty(PASSWORD_PROPERTY, "user1");
+ }
+
+ public static boolean insecure() {
+ return Boolean.parseBoolean(System.getProperty(INSECURE_PROPERTY, "true"));
+ }
+}
diff --git a/src/test/java/com/streamx/cli/commands/auth/KeycloakDeviceApprover.java b/src/test/java/com/streamx/cli/commands/auth/KeycloakDeviceApprover.java
new file mode 100644
index 00000000..53373f51
--- /dev/null
+++ b/src/test/java/com/streamx/cli/commands/auth/KeycloakDeviceApprover.java
@@ -0,0 +1,144 @@
+package com.streamx.cli.commands.auth;
+
+import java.io.IOException;
+import java.net.URI;
+import java.nio.charset.StandardCharsets;
+import java.security.GeneralSecurityException;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import javax.net.ssl.SSLContext;
+import org.apache.http.NameValuePair;
+import org.apache.http.client.entity.UrlEncodedFormEntity;
+import org.apache.http.client.methods.CloseableHttpResponse;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.client.methods.HttpPost;
+import org.apache.http.client.protocol.HttpClientContext;
+import org.apache.http.conn.ssl.NoopHostnameVerifier;
+import org.apache.http.conn.ssl.TrustAllStrategy;
+import org.apache.http.impl.client.BasicCookieStore;
+import org.apache.http.impl.client.CloseableHttpClient;
+import org.apache.http.impl.client.HttpClientBuilder;
+import org.apache.http.impl.client.HttpClients;
+import org.apache.http.impl.client.LaxRedirectStrategy;
+import org.apache.http.message.BasicNameValuePair;
+import org.apache.http.ssl.SSLContexts;
+import org.apache.http.util.EntityUtils;
+
+public class KeycloakDeviceApprover implements AutoCloseable {
+ private static final Pattern FORM_ACTION = Pattern.compile("]*action=\"([^\"]+)\"");
+ private static final Pattern HIDDEN_INPUT = Pattern.compile(
+ "]*type=\"hidden\"[^>]*name=\"([^\"]+)\"[^>]*value=\"([^\"]*)\"");
+
+ private final CloseableHttpClient httpClient;
+ private final HttpClientContext context = HttpClientContext.create();
+
+ public KeycloakDeviceApprover(boolean insecure) {
+ this.context.setCookieStore(new BasicCookieStore());
+
+ HttpClientBuilder builder = HttpClients.custom()
+ .setRedirectStrategy(new LaxRedirectStrategy());
+
+ if (insecure) {
+ try {
+ SSLContext sslContext = SSLContexts.custom()
+ .loadTrustMaterial(null, TrustAllStrategy.INSTANCE)
+ .build();
+ builder.setSSLContext(sslContext).setSSLHostnameVerifier(NoopHostnameVerifier.INSTANCE);
+ } catch (GeneralSecurityException e) {
+ throw new IllegalStateException("Cannot build insecure test http client", e);
+ }
+ }
+ this.httpClient = builder.build();
+ }
+
+ public void approve(String verificationUriComplete, String username, String password)
+ throws IOException {
+ Page page = get(verificationUriComplete);
+
+ if (page.action == null) {
+ throw new IllegalStateException("No form on the device verification page: " + page.summary());
+ }
+
+ if (page.action.contains("login-actions") || page.action.contains("authenticate")) {
+ Map credentials = new LinkedHashMap<>(page.hiddenFields);
+ credentials.put("username", username);
+ credentials.put("password", password);
+ page = post(page.action, credentials);
+ }
+
+ if (page.action == null || !page.action.contains("consent")) {
+ throw new IllegalStateException(
+ "Expected a consent form after login, got: " + page.summary());
+ }
+
+ Map consent = new LinkedHashMap<>(page.hiddenFields);
+ consent.put("accept", "Yes");
+ post(page.action, consent);
+ }
+
+ private Page get(String url) throws IOException {
+ HttpGet request = new HttpGet(url);
+ request.setHeader("User-Agent", "Mozilla/5.0");
+ return execute(request, url);
+ }
+
+ private Page post(String url, Map form) throws IOException {
+ HttpPost request = new HttpPost(url);
+ request.setHeader("User-Agent", "Mozilla/5.0");
+ List params = new ArrayList<>();
+ form.forEach((k, v) -> params.add(new BasicNameValuePair(k, v)));
+ request.setEntity(new UrlEncodedFormEntity(params, StandardCharsets.UTF_8));
+ return execute(request, url);
+ }
+
+ private Page execute(org.apache.http.client.methods.HttpUriRequest request, String requestUrl)
+ throws IOException {
+ try (CloseableHttpResponse response = httpClient.execute(request, context)) {
+ String body = response.getEntity() == null
+ ? ""
+ : EntityUtils.toString(response.getEntity(), StandardCharsets.UTF_8);
+ return new Page(body, resolveCurrentUrl(requestUrl));
+ }
+ }
+
+ private String resolveCurrentUrl(String requestUrl) {
+ List redirects = context.getRedirectLocations();
+ if (redirects != null && !redirects.isEmpty()) {
+ return redirects.get(redirects.size() - 1).toString();
+ }
+ return requestUrl;
+ }
+
+ @Override
+ public void close() throws IOException {
+ httpClient.close();
+ }
+
+ private static final class Page {
+ private final String body;
+ private final String action;
+ private final Map hiddenFields = new LinkedHashMap<>();
+
+ private Page(String body, String currentUrl) {
+ this.body = body;
+
+ Matcher formMatcher = FORM_ACTION.matcher(body);
+ this.action = formMatcher.find()
+ ? URI.create(currentUrl).resolve(formMatcher.group(1).replace("&", "&")).toString()
+ : null;
+
+ Matcher hiddenMatcher = HIDDEN_INPUT.matcher(body);
+ while (hiddenMatcher.find()) {
+ hiddenFields.put(hiddenMatcher.group(1), hiddenMatcher.group(2));
+ }
+ }
+
+ private String summary() {
+ return body.replaceAll("<[^>]+>", " ").replaceAll("\\s+", " ").strip();
+ }
+ }
+}
diff --git a/src/test/java/com/streamx/cli/commands/auth/StubOidcServer.java b/src/test/java/com/streamx/cli/commands/auth/StubOidcServer.java
new file mode 100644
index 00000000..6f8f3c6f
--- /dev/null
+++ b/src/test/java/com/streamx/cli/commands/auth/StubOidcServer.java
@@ -0,0 +1,255 @@
+package com.streamx.cli.commands.auth;
+
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpServer;
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Base64;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class StubOidcServer implements AutoCloseable {
+ public static final String USER_CODE = "WDJB-MJHT";
+ public static final String DEVICE_CODE = "test-device-code";
+ public static final String REFRESH_TOKEN = "test-refresh-token";
+
+ public static final String USERNAME = "user1";
+ public static final String EMAIL = "user1@streamx.com";
+ public static final String SUBJECT = "0a084fdd-7b0c-4ee0-821d-37c78fb43c09";
+ public static final String AUTH_CODE = "test-auth-code";
+
+ public static final String ACCESS_TOKEN = unsignedJwt("""
+ {"preferred_username":"%s","name":"User First","email":"%s","sub":"%s",
+ "iss":"https://keycloak.example/realms/streamx","azp":"streamx-cli"}
+ """.formatted(USERNAME, EMAIL, SUBJECT));
+
+ private static String unsignedJwt(String claims) {
+ Base64.Encoder encoder = Base64.getUrlEncoder().withoutPadding();
+ return encoder.encodeToString("{\"alg\":\"none\"}".getBytes(StandardCharsets.UTF_8))
+ + "." + encoder.encodeToString(claims.getBytes(StandardCharsets.UTF_8))
+ + ".signature";
+ }
+
+ private final HttpServer server;
+ private final String realm;
+
+ private final int pendingResponses;
+
+ private final AtomicInteger tokenRequests = new AtomicInteger();
+ private final List revokedTokens = new ArrayList<>();
+
+ private volatile String tokenError;
+ private volatile int tokenStatus;
+ private volatile String tokenBody = "";
+ private volatile boolean authorizationDenied;
+ private volatile boolean wrongIssuer;
+ private volatile boolean offlineRequested;
+ private volatile boolean offlineScopeRefused;
+ private volatile boolean revocationFails;
+ private volatile String lastRequestedScope = "";
+ private volatile String lastTokenRequestBody = "";
+ private final Map lastAuthorizationRequest = new java.util.HashMap<>();
+
+ public StubOidcServer(String realm, int pendingResponses) throws IOException {
+ this.realm = realm;
+ this.pendingResponses = pendingResponses;
+ this.server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
+
+ String base = "/realms/" + realm;
+ server.createContext(base + "/.well-known/openid-configuration", this::handleDiscovery);
+ server.createContext(base + "/auth", this::handleAuthorization);
+ server.createContext(base + "/device", this::handleDeviceAuthorization);
+ server.createContext(base + "/token", this::handleToken);
+ server.createContext(base + "/revoke", this::handleRevoke);
+ server.start();
+ }
+
+ public String getServerUrl() {
+ return "http://127.0.0.1:" + server.getAddress().getPort();
+ }
+
+ public int getTokenRequestCount() {
+ return tokenRequests.get();
+ }
+
+ public List getRevokedTokens() {
+ return revokedTokens;
+ }
+
+ public void failTokenWith(String error) {
+ this.tokenError = error;
+ }
+
+ public void failTokenWithStatus(int status, String body) {
+ this.tokenStatus = status;
+ this.tokenBody = body;
+ }
+
+ public void denyAuthorization() {
+ this.authorizationDenied = true;
+ }
+
+ public void returnWrongIssuer() {
+ this.wrongIssuer = true;
+ }
+
+ public void refuseOfflineScope() {
+ this.offlineScopeRefused = true;
+ }
+
+ public void failRevocation() {
+ this.revocationFails = true;
+ }
+
+ public Map getLastAuthorizationRequest() {
+ return lastAuthorizationRequest;
+ }
+
+ public String getLastTokenRequestBody() {
+ return lastTokenRequestBody;
+ }
+
+ public String getLastRequestedScope() {
+ return lastRequestedScope;
+ }
+
+ private static Map parseQuery(String rawQuery) {
+ Map params = new java.util.HashMap<>();
+ if (rawQuery == null) {
+ return params;
+ }
+ for (String pair : rawQuery.split("&")) {
+ int eq = pair.indexOf('=');
+ if (eq > 0) {
+ params.put(
+ java.net.URLDecoder.decode(pair.substring(0, eq), StandardCharsets.UTF_8),
+ java.net.URLDecoder.decode(pair.substring(eq + 1), StandardCharsets.UTF_8));
+ }
+ }
+ return params;
+ }
+
+ private void handleDiscovery(HttpExchange exchange) throws IOException {
+ String base = getServerUrl() + "/realms/" + realm;
+ String issuer = wrongIssuer ? "https://evil.example/realms/" + realm : base;
+ respond(exchange, 200, """
+ {
+ "issuer": "%s",
+ "authorization_endpoint": "%s/auth",
+ "device_authorization_endpoint": "%s/device",
+ "token_endpoint": "%s/token",
+ "revocation_endpoint": "%s/revoke"
+ }
+ """.formatted(issuer, base, base, base, base));
+ }
+
+ private void handleAuthorization(HttpExchange exchange) throws IOException {
+ Map query = parseQuery(exchange.getRequestURI().getRawQuery());
+ lastAuthorizationRequest.putAll(query);
+ recordScope(exchange.getRequestURI().getRawQuery());
+
+ String redirectUri = query.get("redirect_uri");
+ String state = query.get("state");
+ String location = authorizationDenied
+ ? redirectUri + "?error=access_denied&state=" + state
+ : redirectUri + "?code=" + AUTH_CODE + "&state=" + state;
+
+ exchange.getResponseHeaders().add("Location", location);
+ exchange.sendResponseHeaders(302, -1);
+ exchange.close();
+ }
+
+ private void handleDeviceAuthorization(HttpExchange exchange) throws IOException {
+ recordScope(readBody(exchange));
+ respond(exchange, 200, """
+ {
+ "device_code": "%s",
+ "user_code": "%s",
+ "verification_uri": "%s/device",
+ "verification_uri_complete": "%s/device?user_code=%s",
+ "interval": 1,
+ "expires_in": 60
+ }
+ """.formatted(DEVICE_CODE, USER_CODE, getServerUrl(), getServerUrl(), USER_CODE));
+ }
+
+ private void handleToken(HttpExchange exchange) throws IOException {
+ lastTokenRequestBody = readBody(exchange);
+ int attempt = tokenRequests.incrementAndGet();
+
+ if (tokenStatus != 0) {
+ respond(exchange, tokenStatus, tokenBody);
+ return;
+ }
+ if (tokenError != null) {
+ respond(exchange, 400, "{\"error\": \"%s\"}".formatted(tokenError));
+ return;
+ }
+ if (attempt <= pendingResponses) {
+ respond(exchange, 400, "{\"error\": \"authorization_pending\"}");
+ return;
+ }
+
+ String refreshToken = offlineRequested ? getOfflineToken() : REFRESH_TOKEN;
+ respond(exchange, 200, """
+ {
+ "access_token": "%s",
+ "refresh_token": "%s",
+ "expires_in": 300,
+ "token_type": "Bearer"
+ }
+ """.formatted(ACCESS_TOKEN, refreshToken));
+ }
+
+ public String getOfflineToken() {
+ return unsignedJwt("""
+ {"typ":"Offline","iss":"%s/realms/%s","azp":"streamx-cli","sub":"%s"}
+ """.formatted(getServerUrl(), realm, SUBJECT));
+ }
+
+ private void recordScope(String formOrQuery) {
+ String scope = parseQuery(formOrQuery).getOrDefault("scope", "");
+ lastRequestedScope = scope;
+ if (scope.contains("offline_access") && !offlineScopeRefused) {
+ offlineRequested = true;
+ }
+ }
+
+ private void handleRevoke(HttpExchange exchange) throws IOException {
+ String body = readBody(exchange);
+ if (revocationFails) {
+ respond(exchange, 503, "{\"error\":\"temporarily_unavailable\"}");
+ return;
+ }
+ for (String param : body.split("&")) {
+ if (param.startsWith("token=")) {
+ revokedTokens.add(
+ java.net.URLDecoder.decode(param.substring("token=".length()), StandardCharsets.UTF_8));
+ }
+ }
+ respond(exchange, 200, "{}");
+ }
+
+ private static String readBody(HttpExchange exchange) throws IOException {
+ try (InputStream in = exchange.getRequestBody()) {
+ return new String(in.readAllBytes(), StandardCharsets.UTF_8);
+ }
+ }
+
+ private static void respond(HttpExchange exchange, int status, String body) throws IOException {
+ byte[] bytes = body.getBytes(StandardCharsets.UTF_8);
+ exchange.getResponseHeaders().add("Content-Type", "application/json");
+ exchange.sendResponseHeaders(status, bytes.length);
+ exchange.getResponseBody().write(bytes);
+ exchange.close();
+ }
+
+ @Override
+ public void close() {
+ server.stop(0);
+ }
+}
diff --git a/src/test/java/com/streamx/cli/commands/auth/TestHttpClient.java b/src/test/java/com/streamx/cli/commands/auth/TestHttpClient.java
new file mode 100644
index 00000000..f18f6b3c
--- /dev/null
+++ b/src/test/java/com/streamx/cli/commands/auth/TestHttpClient.java
@@ -0,0 +1,54 @@
+package com.streamx.cli.commands.auth;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.security.GeneralSecurityException;
+import javax.net.ssl.SSLContext;
+import org.apache.http.client.methods.CloseableHttpResponse;
+import org.apache.http.client.methods.HttpPost;
+import org.apache.http.conn.ssl.NoopHostnameVerifier;
+import org.apache.http.conn.ssl.TrustAllStrategy;
+import org.apache.http.entity.StringEntity;
+import org.apache.http.impl.client.CloseableHttpClient;
+import org.apache.http.impl.client.HttpClients;
+import org.apache.http.ssl.SSLContexts;
+import org.apache.http.util.EntityUtils;
+
+public class TestHttpClient implements AutoCloseable {
+ private final CloseableHttpClient httpClient;
+
+ public TestHttpClient(boolean insecure) {
+ if (!insecure) {
+ this.httpClient = HttpClients.custom().build();
+ return;
+ }
+ try {
+ SSLContext sslContext = SSLContexts.custom()
+ .loadTrustMaterial(null, TrustAllStrategy.INSTANCE)
+ .build();
+ this.httpClient = HttpClients.custom()
+ .setSSLContext(sslContext)
+ .setSSLHostnameVerifier(NoopHostnameVerifier.INSTANCE)
+ .build();
+ } catch (GeneralSecurityException e) {
+ throw new IllegalStateException("Cannot build insecure test http client", e);
+ }
+ }
+
+ public String postForm(String url, String formBody) throws IOException {
+ HttpPost request = new HttpPost(url);
+ request.setHeader("Content-Type", "application/x-www-form-urlencoded");
+ request.setEntity(new StringEntity(formBody, StandardCharsets.UTF_8));
+
+ try (CloseableHttpResponse response = httpClient.execute(request)) {
+ return response.getEntity() == null
+ ? ""
+ : EntityUtils.toString(response.getEntity(), StandardCharsets.UTF_8);
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ httpClient.close();
+ }
+}
diff --git a/src/test/java/com/streamx/cli/commands/context/ContextCommandIT.java b/src/test/java/com/streamx/cli/commands/context/ContextCommandIT.java
index 7b5175b8..8e02c944 100644
--- a/src/test/java/com/streamx/cli/commands/context/ContextCommandIT.java
+++ b/src/test/java/com/streamx/cli/commands/context/ContextCommandIT.java
@@ -4,6 +4,7 @@
import static com.streamx.cli.i18n.MessageProvider.msg;
import static org.assertj.core.api.Assertions.assertThat;
+import com.streamx.cli.commands.auth.StubOidcServer;
import com.streamx.cli.test.CliBaseIT;
import io.quarkus.test.junit.QuarkusTest;
import java.io.IOException;
@@ -11,12 +12,14 @@
import java.nio.file.Path;
import java.util.Comparator;
import java.util.stream.Stream;
+import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@QuarkusTest
class ContextCommandIT extends CliBaseIT {
+ private StubOidcServer oidcServer;
@BeforeEach
void cleanContexts() throws IOException {
@@ -24,6 +27,14 @@ void cleanContexts() throws IOException {
Files.deleteIfExists(streamxHome.resolve("current-context"));
}
+ @AfterEach
+ void stopServer() {
+ if (oidcServer != null) {
+ oidcServer.close();
+ oidcServer = null;
+ }
+ }
+
@Test
void firstUseBootstrapsDefaultContext() throws Exception {
ProcessResult result = exec("context", "current");
@@ -207,6 +218,23 @@ void globalFlagsWorkAtAnyPosition() throws Exception {
assertThat(altHome.resolve("contexts/default/config")).isDirectory();
}
+ @Test
+ void loginWritesIntoTheActiveContext() throws Exception {
+ oidcServer = new StubOidcServer("streamx", 0);
+ exec("context", "create", "prod").assertSuccess();
+ exec("context", "use", "prod").assertSuccess();
+ Files.writeString(streamxHome.resolve("contexts/prod/config/application.properties"), """
+ streamx.auth.server-url=%s
+ streamx.auth.realm=streamx
+ streamx.auth.client-id=streamx-cli
+ """.formatted(oidcServer.getServerUrl()));
+
+ exec("auth", "login", "--no-browser").assertSuccess();
+
+ assertThat(streamxHome.resolve("contexts/prod/config/credentials.json")).exists();
+ assertThat(streamxHome.resolve("contexts/default/config/credentials.json")).doesNotExist();
+ }
+
@Test
void helpHeaderShowsCurrentContext() throws Exception {
ProcessResult defaultHelp = exec("--help");
diff --git a/src/test/java/com/streamx/cli/commands/info/InfoCommandIT.java b/src/test/java/com/streamx/cli/commands/info/InfoCommandIT.java
new file mode 100644
index 00000000..e5847842
--- /dev/null
+++ b/src/test/java/com/streamx/cli/commands/info/InfoCommandIT.java
@@ -0,0 +1,186 @@
+package com.streamx.cli.commands.info;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.streamx.cli.commands.auth.StubOidcServer;
+import com.streamx.cli.platform.AccessTokens;
+import com.streamx.cli.test.CliBaseIT;
+import io.quarkus.test.junit.QuarkusTest;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.time.Instant;
+import java.util.Base64;
+import java.util.Comparator;
+import java.util.stream.Stream;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+@QuarkusTest
+class InfoCommandIT extends CliBaseIT {
+
+ private static final ObjectMapper JSON = new ObjectMapper();
+
+ private StubOidcServer oidcServer;
+
+ @BeforeEach
+ void cleanContexts() throws IOException {
+ deleteRecursively(streamxHome.resolve("contexts"));
+ Files.deleteIfExists(streamxHome.resolve("current-context"));
+ }
+
+ @AfterEach
+ void stopServer() {
+ clearEnv(AccessTokens.STREAMX_PLATFORM_TOKEN);
+ if (oidcServer != null) {
+ oidcServer.close();
+ oidcServer = null;
+ }
+ }
+
+ @Test
+ void worksOnFreshHomeWithoutAnyConfiguration() throws Exception {
+ ProcessResult result = exec("info");
+
+ result.assertSuccess();
+ assertThat(result.stdout())
+ .contains("CLI")
+ .contains("active")
+ .contains("default")
+ .contains("not logged in")
+ .contains("(no endpoints configured)")
+ .contains("Auth server is not configured");
+ }
+
+ @Test
+ void probesConfiguredEndpointsAndReportsSources() throws Exception {
+ oidcServer = new StubOidcServer("streamx", 0);
+ exec("settings", "set", "streamx.auth.server-url", oidcServer.getServerUrl())
+ .assertSuccess();
+ exec("settings", "set", "streamx.platform.url", oidcServer.getServerUrl())
+ .assertSuccess();
+
+ ProcessResult result = exec("info");
+
+ result.assertSuccess();
+ assertThat(result.stdout())
+ .contains("streamx.auth.server-url")
+ .contains("context")
+ .contains("discovery document served")
+ // The stub answers HTTP but 404s the API routes: reachable yet not the platform.
+ .contains("not the StreamX platform API");
+
+ ProcessResult json = exec("info", "-o", "json");
+ json.assertSuccess();
+ JsonNode root = JSON.readTree(json.stdout());
+ assertThat(root.path("cli").path("version").asText()).isNotEmpty();
+ assertThat(root.path("connectivity").isArray()).isTrue();
+ assertThat(root.path("connectivity").toString()).contains("\"UP\"");
+ }
+
+ @Test
+ void checkFlagFailsWhenPlatformAnswersButIsNotThePlatform() throws Exception {
+ oidcServer = new StubOidcServer("streamx", 0);
+ exec("settings", "set", "streamx.platform.url", oidcServer.getServerUrl())
+ .assertSuccess();
+
+ ProcessResult result = exec("info", "--check");
+
+ result.assertExitCode(1);
+ assertThat(result.stdout()).contains("HTTP 404").contains("DOWN");
+ }
+
+ @Test
+ void checkFlagFailsWhenAnEndpointIsDown() throws Exception {
+ exec("settings", "set", "streamx.platform.url", "https://127.0.0.1:1").assertSuccess();
+
+ ProcessResult result = exec("info", "--check");
+
+ result.assertExitCode(1);
+ assertThat(result.stdout()).contains("DOWN");
+ }
+
+ @Test
+ void reportsStoredLoginAndIssuerMismatch() throws Exception {
+ oidcServer = new StubOidcServer("streamx", 0);
+ exec("settings", "set", "streamx.auth.server-url", oidcServer.getServerUrl())
+ .assertSuccess();
+
+ String payload = Base64.getUrlEncoder().withoutPadding().encodeToString(
+ "{\"preferred_username\":\"tester\"}".getBytes(StandardCharsets.UTF_8));
+ String header = Base64.getUrlEncoder().withoutPadding().encodeToString(
+ "{\"alg\":\"none\"}".getBytes(StandardCharsets.UTF_8));
+ String credentials = """
+ {
+ "access_token": "%s.%s.sig",
+ "refresh_token": "r",
+ "expires_at": %d,
+ "issuer_url": "https://other-idp.example.com/realms/streamx",
+ "client_id": "streamx-cli",
+ "insecure": false
+ }
+ """.formatted(header, payload, Instant.now().plusSeconds(3600).getEpochSecond());
+ Files.writeString(
+ streamxHome.resolve("contexts/default/config/credentials.json"), credentials);
+
+ ProcessResult result = exec("info");
+
+ result.assertSuccess();
+ assertThat(result.stdout())
+ .contains("logged in")
+ .contains("tester")
+ .contains("Stored login belongs to https://other-idp.example.com/realms/streamx");
+ }
+
+ @Test
+ void refusesToProbeWithACredentialOverCleartextHttp() throws Exception {
+ exec("settings", "set", "streamx.platform.url", "http://platform.example.com").assertSuccess();
+ setEnv(AccessTokens.STREAMX_PLATFORM_TOKEN, "sxp_v1_token");
+
+ ProcessResult result = exec("info");
+
+ result.assertSuccess();
+ assertThat(result.stdout()).contains("refusing to send a credential over cleartext http");
+ }
+
+ @Test
+ void stillReportsWhenTheStoredSessionCannotBeRefreshed() throws Exception {
+ exec("settings", "set", "streamx.platform.url", "https://127.0.0.1:1").assertSuccess();
+ String credentials = """
+ {
+ "access_token": "stale",
+ "refresh_token": "r",
+ "expires_at": %d,
+ "issuer_url": "http://127.0.0.1:1/realms/streamx",
+ "client_id": "streamx-cli"
+ }
+ """.formatted(Instant.now().minusSeconds(3600).getEpochSecond());
+ Files.createDirectories(streamxHome.resolve("contexts/default/config"));
+ Files.writeString(
+ streamxHome.resolve("contexts/default/config/credentials.json"), credentials);
+
+ ProcessResult result = exec("info");
+
+ result.assertSuccess();
+ assertThat(result.stdout()).contains("no usable credential");
+ }
+
+ private static void deleteRecursively(Path root) throws IOException {
+ if (!Files.exists(root)) {
+ return;
+ }
+ try (Stream paths = Files.walk(root)) {
+ paths.sorted(Comparator.reverseOrder()).forEach(path -> {
+ try {
+ Files.delete(path);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ });
+ }
+ }
+}
diff --git a/src/test/java/com/streamx/cli/platform/PlatformClientsCleartextTest.java b/src/test/java/com/streamx/cli/platform/PlatformClientsCleartextTest.java
new file mode 100644
index 00000000..b3d30776
--- /dev/null
+++ b/src/test/java/com/streamx/cli/platform/PlatformClientsCleartextTest.java
@@ -0,0 +1,24 @@
+package com.streamx.cli.platform;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import com.streamx.cli.framework.CliException;
+import org.junit.jupiter.api.Test;
+
+class PlatformClientsCleartextTest {
+
+ @Test
+ void refusesCleartextRemotePlatformUrl() {
+ assertThatThrownBy(() -> new PlatformClients("http://platform.example.com", false, 30_000))
+ .isInstanceOf(CliException.class)
+ .hasMessageContaining("cleartext HTTP");
+ }
+
+ @Test
+ void allowsLoopbackHttpForLocalDevelopment() {
+ try (PlatformClients clients = new PlatformClients("http://localhost:8085", false, 30_000)) {
+ assertThat(clients).isNotNull();
+ }
+ }
+}