[api][runtime][java][python] Require HTTPS and add digest pinning for URL skill sources

api/src/main/java/org/apache/flink/agents/api/skills/Skills.java

@@ -25,9 +25,11 @@
 import org.apache.flink.agents.api.resource.ResourceType;
 import org.apache.flink.agents.api.resource.SerializableResource;
 
+import java.net.URI;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import java.util.stream.Collectors;
 
@@ -41,7 +43,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>
  *
@@ -92,17 +94,94 @@ 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 -> {
+                                    requireUrl(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 -> {
+                                    requireUrl(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) {
+        requireUrl(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)));
+    }
+
+    private static void requireUrl(String url, boolean allowInsecureHttp) {
+        if (url == null) {
+            throw new IllegalArgumentException("skill URL must not be null");
+        }
+        URI uri;
+        try {
+            uri = URI.create(url);
+        } catch (IllegalArgumentException e) {
+            throw new IllegalArgumentException("Invalid skill URL: " + url, e);
+        }
+        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: " + 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: "
+                            + url);
+        }
+        if (uri.getRawAuthority() == null || uri.getRawAuthority().isEmpty()) {
+            throw new IllegalArgumentException("Skill URL must include a host: " + url);
+        }
+    }
+
     /**
      * Create a {@link Skills} resource from one or more classpath resource paths.
      *

api/src/main/java/org/apache/flink/agents/api/yaml/YamlLoader.java

@@ -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;
@@ -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)));

api/src/main/java/org/apache/flink/agents/api/yaml/spec/SkillsSpec.java

@@ -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;
 
@@ -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.");
         }
     }
 
@@ -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;
     }

api/src/main/java/org/apache/flink/agents/api/yaml/spec/UrlSkillSpec.java

@@ -0,0 +1,57 @@
+/*
+ * 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.yaml.spec;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import javax.annotation.Nullable;
+
+/** Declarative configuration for one URL-backed skill archive. */
+@JsonIgnoreProperties(ignoreUnknown = false)
+public final class UrlSkillSpec {
+    private final String url;
+    @Nullable private final String sha256;
+    private final boolean allowInsecureHttp;
+
+    @JsonCreator
+    public UrlSkillSpec(
+            @JsonProperty(value = "url", required = true) String url,
+            @JsonProperty("sha256") @Nullable String sha256,
+            @JsonProperty("allow_insecure_http") boolean allowInsecureHttp) {
+        this.url = url;
+        this.sha256 = sha256;
+        this.allowInsecureHttp = allowInsecureHttp;
+    }
+
+    public String getUrl() {
+        return url;
+    }
+
+    @Nullable
+    public String getSha256() {
+        return sha256;
+    }
+
+    @JsonProperty("allow_insecure_http")
+    public boolean isAllowInsecureHttp() {
+        return allowInsecureHttp;
+    }
+}

api/src/test/java/org/apache/flink/agents/api/skills/SkillsResourceTest.java

@@ -18,6 +18,7 @@
 
 package org.apache.flink.agents.api.skills;
 
+import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import org.apache.flink.agents.api.resource.ResourceType;
 import org.junit.jupiter.api.Test;
@@ -26,6 +27,8 @@
 import java.util.Map;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 
 class SkillsResourceTest {
 
@@ -48,6 +51,62 @@ void fromUrlEmitsUrlScheme() {
                 skills.getSources());
     }
 
+    @Test
+    void fromUrlWithSha256EmitsIntegrityParam() {
+        String digest = "a".repeat(64);
+        Skills skills = Skills.fromUrlWithSha256("https://example.com/x.zip", digest);
+        assertEquals(
+                List.of(
+                        new SkillSourceSpec(
+                                "url",
+                                Map.of("url", "https://example.com/x.zip", "sha256", digest))),
+                skills.getSources());
+    }
+
+    @Test
+    void fromUrlUnsafeRequiresExplicitParam() {
+        Skills skills = Skills.fromUrlUnsafe("http://example.com/x.zip");
+        assertEquals("true", skills.getSources().get(0).getParams().get("allow_insecure_http"));
+    }
+
+    @Test
+    void fromUrlUnsafeWithSha256EmitsBothParams() {
+        String digest = "a".repeat(64);
+        Skills skills = Skills.fromUrlUnsafeWithSha256("http://example.com/x.zip", digest);
+        assertEquals(
+                Map.of(
+                        "url",
+                        "http://example.com/x.zip",
+                        "sha256",
+                        digest,
+                        "allow_insecure_http",
+                        "true"),
+                skills.getSources().get(0).getParams());
+    }
+
+    @Test
+    void fromUrlRejectsPlainHttpByDefault() {
+        assertThrows(
+                IllegalArgumentException.class, () -> Skills.fromUrl("http://example.com/x.zip"));
+    }
+
+    @Test
+    void fromUrlWithSha256RejectsMalformedDigest() {
+        assertThrows(
+                IllegalArgumentException.class,
+                () -> Skills.fromUrlWithSha256("https://example.com/x.zip", "invalid"));
+    }
+
+    @Test
+    void fromUrlRejectsUnsupportedSchemeClearly() {
+        IllegalArgumentException ex =
+                assertThrows(
+                        IllegalArgumentException.class,
+                        () -> Skills.fromUrl("ftp://example.com/x.zip"));
+        assertEquals(
+                "Only HTTP(S) skill URLs are supported: ftp://example.com/x.zip", ex.getMessage());
+    }
+
     @Test
     void fromClasspathEmitsClasspathScheme() {
         Skills skills = Skills.fromClasspath("skills");
@@ -65,6 +124,20 @@ void roundTripsThroughJackson() throws Exception {
         assertEquals(original.getSources(), restored.getSources());
     }
 
+    @Test
+    void unsafePinnedUrlRoundTripsThroughJackson() throws Exception {
+        Skills original =
+                Skills.fromUrlUnsafeWithSha256("http://example.com/skills.zip", "a".repeat(64));
+        ObjectMapper mapper = new ObjectMapper();
+        String json = mapper.writeValueAsString(original);
+        JsonNode allowInsecureHttp =
+                mapper.readTree(json).at("/sources/0/params/allow_insecure_http");
+        assertTrue(allowInsecureHttp.isTextual());
+        assertEquals("true", allowInsecureHttp.asText());
+        Skills restored = mapper.readValue(json, Skills.class);
+        assertEquals(original.getSources(), restored.getSources());
+    }
+
     @Test
     void reservedNamesMatchPython() {
         assertEquals("_skills_config", Skills.SKILLS_CONFIG);