Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
@@ -0,0 +1,188 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/

package org.apache.flink.agents.api.skills;

import org.apache.flink.annotation.Internal;

import java.net.MalformedURLException;
import java.net.URI;
import java.net.URISyntaxException;
import java.net.URL;
import java.util.Locale;
import java.util.regex.Pattern;

/**
* Shared validation and redaction helpers for URL-backed skill sources. Internal contract shared
* with the runtime module; not a stable public API.
*/
@Internal
public final class SkillUrlUtils {

private static final String REDACTED = "<redacted>";
private static final Pattern INVALID_PERCENT_ESCAPE = Pattern.compile("%(?![0-9a-fA-F]{2})");

private SkillUrlUtils() {}

/**
* Validate {@code url} and return its lowercase {@code http} or {@code https} scheme.
*
* @throws IllegalArgumentException if the URL is invalid or violates the transport policy.
*/
public static String validate(String url, boolean allowInsecureHttp) {
if (url == null) {
throw new IllegalArgumentException("skill URL must not be null");
}
if (INVALID_PERCENT_ESCAPE.matcher(url).find()) {
throw new IllegalArgumentException("Invalid skill URL: " + redact(url));
}
URI uri;
try {
uri = URI.create(url);
} catch (IllegalArgumentException ignored) {
throw new IllegalArgumentException("Invalid skill URL: " + redact(url));
}
String scheme = uri.getScheme();
scheme = scheme == null ? "" : scheme.toLowerCase(Locale.ROOT);
if (!(scheme.equals("http") || scheme.equals("https"))) {
throw new IllegalArgumentException(
"Only HTTP(S) skill URLs are supported: " + redact(url));
}
try {
uri = uri.parseServerAuthority();
} catch (URISyntaxException ignored) {
throw new IllegalArgumentException(
"Skill URL must include a valid host and, when present, a valid port: "
+ redact(url));
}
if (uri.getRawUserInfo() != null) {
throw new IllegalArgumentException(
"Skill URL must not include user info: " + redact(url));
}
String host = uri.getHost();
if (host == null || host.isEmpty()) {
throw new IllegalArgumentException(
"Skill URL must include a valid host: " + redact(url));
}
if (host.indexOf(':') >= 0 && host.indexOf('%') >= 0) {
throw new IllegalArgumentException(
"Skill URL must not include an IPv6 zone identifier: " + redact(url));
}
if (uri.getPort() > 65535) {
throw new IllegalArgumentException(
"Skill URL port must be between 0 and 65535: " + redact(url));
}
if (scheme.equals("http") && !allowInsecureHttp) {
throw new IllegalArgumentException(
"Plain HTTP skill URLs are disabled by default; use HTTPS or explicitly allow"
+ " insecure HTTP for this source: "
+ redact(url));
}
return scheme;
}

/** Return {@code url} without user info, query parameters, or a fragment. */
public static String redact(String url) {
if (url == null) {
return REDACTED;
}
try {
URI uri = URI.create(url);
if (uri.getScheme() == null || uri.getRawAuthority() == null) {
return REDACTED;
}
return redactParts(uri.getScheme(), uri.getRawAuthority(), uri.getRawPath());
} catch (IllegalArgumentException ignored) {
try {
URL parsed = new URL(url);
String authority = parsed.getAuthority();
String path = parsed.getPath();
if (authority == null
|| containsUnsafeLogCharacter(authority)
|| containsUnsafeLogCharacter(path)) {
return REDACTED;
}
return redactParts(parsed.getProtocol(), authority, path);
} catch (MalformedURLException | IllegalArgumentException malformed) {
return REDACTED;
}
}
}

private static String redactParts(String scheme, String authority, String path) {
int userInfoEnd = authority.lastIndexOf('@');
if (userInfoEnd >= 0) {
authority = authority.substring(userInfoEnd + 1);
}
if (authority.isEmpty() || hasInvalidPort(authority)) {
return REDACTED;
}
return scheme + "://" + authority + (path == null ? "" : path);
}

private static boolean hasInvalidPort(String authority) {
int portStart;
if (authority.startsWith("[")) {
int closingBracket = authority.lastIndexOf(']');
if (closingBracket < 0) {
return true;
}
if (closingBracket == authority.length() - 1) {
return false;
}
if (authority.charAt(closingBracket + 1) != ':') {
return true;
}
portStart = closingBracket + 2;
} else {
int firstColon = authority.indexOf(':');
if (firstColon < 0) {
return false;
}
if (firstColon != authority.lastIndexOf(':')) {
return true;
}
portStart = firstColon + 1;
}
if (portStart == authority.length()) {
return false;
}
int port = 0;
for (int i = portStart; i < authority.length(); i++) {
char c = authority.charAt(i);
if (c < '0' || c > '9') {
return true;
}
port = port * 10 + (c - '0');
if (port > 65535) {
return true;
}
}
return false;
}

private static boolean containsUnsafeLogCharacter(String value) {
for (int i = 0; i < value.length(); i++) {
char c = value.charAt(i);
if (c < 0x20 || c == 0x7f) {
return true;
}
}
return false;
}
}
57 changes: 54 additions & 3 deletions api/src/main/java/org/apache/flink/agents/api/skills/Skills.java
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,7 @@
*
* <ul>
* <li>{@link #fromLocalDir(String...)} for local directories or {@code .zip} files
* <li>{@link #fromUrl(String...)} for http(s) URLs pointing to a {@code .zip}
* <li>{@link #fromUrl(String...)} for HTTPS URLs pointing to a {@code .zip}
* <li>{@link #fromClasspath(String...)} for resources on the classpath
* </ul>
*
Expand Down Expand Up @@ -92,17 +92,68 @@ public static Skills fromLocalDir(String... paths) {
}

/**
* Create a {@link Skills} resource from one or more http(s) URLs.
* Create a {@link Skills} resource from one or more HTTPS URLs.
*
* <p>Each URL must point to a {@code .zip} whose top level is the baseDir.
*/
public static Skills fromUrl(String... urls) {
return new Skills(
Arrays.stream(urls)
.map(u -> new SkillSourceSpec("url", Map.of("url", u)))
.map(
u -> {
SkillUrlUtils.validate(u, false);
return new SkillSourceSpec("url", Map.of("url", u));
})
.collect(Collectors.toList()));
}

/**
* Create a {@link Skills} resource from an HTTPS URL pinned to a SHA-256 digest.
*
* <p>The digest is verified against the downloaded archive before extraction.
*/
public static Skills fromUrlWithSha256(String url, String sha256) {
return urlSource(url, sha256, false);
}

/**
* Create a {@link Skills} resource that explicitly permits plain HTTP transport.
*
* <p>This compatibility escape hatch should be used only on trusted networks. Prefer {@link
* #fromUrl(String...)} with HTTPS.
*/
public static Skills fromUrlUnsafe(String... urls) {
return new Skills(
Arrays.stream(urls)
.map(
u -> {
SkillUrlUtils.validate(u, true);
return new SkillSourceSpec(
"url", Map.of("url", u, "allow_insecure_http", "true"));
})
.collect(Collectors.toList()));
}

/**
* Create a digest-pinned {@link Skills} resource that explicitly permits plain HTTP transport.
*/
public static Skills fromUrlUnsafeWithSha256(String url, String sha256) {
return urlSource(url, sha256, true);
}

private static Skills urlSource(String url, String sha256, boolean allowInsecureHttp) {
SkillUrlUtils.validate(url, allowInsecureHttp);
if (sha256 == null || !sha256.matches("[0-9a-fA-F]{64}")) {
throw new IllegalArgumentException(
"sha256 must contain exactly 64 hexadecimal characters");
}
Map<String, String> params =
allowInsecureHttp
? Map.of("url", url, "sha256", sha256, "allow_insecure_http", "true")
: Map.of("url", url, "sha256", sha256);
return new Skills(List.of(new SkillSourceSpec("url", params)));
}

/**
* Create a {@link Skills} resource from one or more classpath resource paths.
*
Expand Down
21 changes: 19 additions & 2 deletions api/src/main/java/org/apache/flink/agents/api/yaml/YamlLoader.java
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,7 @@
import org.apache.flink.agents.api.yaml.spec.PromptSpec;
import org.apache.flink.agents.api.yaml.spec.SkillsSpec;
import org.apache.flink.agents.api.yaml.spec.ToolSpec;
import org.apache.flink.agents.api.yaml.spec.UrlSkillSpec;
import org.apache.flink.agents.api.yaml.spec.YamlAgentsDocument;

import java.io.IOException;
Expand Down Expand Up @@ -173,8 +174,24 @@ public static Skills buildSkills(SkillsSpec spec) {
for (String p : spec.getPaths()) {
sources.add(new SkillSourceSpec("local", Map.of("path", p)));
}
for (String u : spec.getUrls()) {
sources.add(new SkillSourceSpec("url", Map.of("url", u)));
for (String url : spec.getUrls()) {
sources.addAll(Skills.fromUrl(url).getSources());
}
for (UrlSkillSpec urlSpec : spec.getUrlSources()) {
Skills urlSkills;
if (urlSpec.isAllowInsecureHttp()) {
urlSkills =
urlSpec.getSha256() == null
? Skills.fromUrlUnsafe(urlSpec.getUrl())
: Skills.fromUrlUnsafeWithSha256(
urlSpec.getUrl(), urlSpec.getSha256());
} else {
urlSkills =
urlSpec.getSha256() == null
? Skills.fromUrl(urlSpec.getUrl())
: Skills.fromUrlWithSha256(urlSpec.getUrl(), urlSpec.getSha256());
}
sources.addAll(urlSkills.getSources());
}
for (String r : spec.getClasspath()) {
sources.add(new SkillSourceSpec("classpath", Map.of("resource", r)));
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -30,21 +30,24 @@
*
* <ul>
* <li>{@code paths} — {@code local} scheme: directories or {@code .zip} files on the filesystem
* <li>{@code urls} — {@code url} scheme: {@code http(s)} URLs pointing to a {@code .zip}
* <li>{@code urls} — {@code url} scheme: HTTPS URLs pointing to {@code .zip} archives
* <li>{@code url_sources} — {@code url} scheme: archive configurations with optional digest and
* transport policy
* <li>{@code classpath} — {@code classpath} scheme: resource paths on the Java classpath
* <li>{@code package} — {@code package} scheme (Python-only at runtime): {@code (package,
* resource)} pairs pointing at resources inside an installed Python package
* </ul>
*
* <p>At least one of the four lists must be non-empty. {@code package} is exposed on Java for YAML
* schema parity with Python — it deserializes successfully but {@code SkillManager} on Java will
* fail at load time because Java does not register a {@code package} handler.
* <p>At least one source list must be non-empty. {@code package} is exposed on Java for YAML schema
* parity with Python — it deserializes successfully but {@code SkillManager} on Java will fail at
* load time because Java does not register a {@code package} handler.
*/
@JsonIgnoreProperties(ignoreUnknown = false)
public final class SkillsSpec {
private final String name;
private final List<String> paths;
private final List<String> urls;
private final List<UrlSkillSpec> urlSources;
private final List<String> classpath;
private final List<PackageSkillSpec> packageEntries;

Expand All @@ -53,22 +56,26 @@ public SkillsSpec(
@JsonProperty(value = "name", required = true) String name,
@JsonProperty("paths") List<String> paths,
@JsonProperty("urls") List<String> urls,
@JsonProperty("url_sources") List<UrlSkillSpec> urlSources,
@JsonProperty("classpath") List<String> classpath,
@JsonProperty("package") List<PackageSkillSpec> packageEntries) {
this.name = name;
this.paths = paths == null ? Collections.emptyList() : List.copyOf(paths);
this.urls = urls == null ? Collections.emptyList() : List.copyOf(urls);
this.urlSources = urlSources == null ? Collections.emptyList() : List.copyOf(urlSources);
this.classpath = classpath == null ? Collections.emptyList() : List.copyOf(classpath);
this.packageEntries =
packageEntries == null ? Collections.emptyList() : List.copyOf(packageEntries);
if (this.paths.isEmpty()
&& this.urls.isEmpty()
&& this.urlSources.isEmpty()
&& this.classpath.isEmpty()
&& this.packageEntries.isEmpty()) {
throw new IllegalArgumentException(
"skills '"
+ name
+ "': at least one of paths/urls/classpath/package must be non-empty.");
+ "': at least one of paths/urls/url_sources/classpath/package must be"
+ " non-empty.");
}
}

Expand All @@ -84,6 +91,11 @@ public List<String> getUrls() {
return urls;
}

@JsonProperty("url_sources")
public List<UrlSkillSpec> getUrlSources() {
return urlSources;
}

public List<String> getClasspath() {
return classpath;
}
Expand Down
Loading
Loading