Introduce support for subscription auto-verification in websubhub

ballerina/Dependencies.toml

@@ -73,7 +73,7 @@ dependencies = [
 [[package]]
 org = "ballerina"
 name = "http"
-version = "2.13.2"
+version = "2.13.3"
 dependencies = [
 	{org = "ballerina", name = "auth"},
 	{org = "ballerina", name = "cache"},

ballerina/annotation.bal

@@ -16,11 +16,13 @@
 
 # Configuration for a WebSub Hub service.
 #
-# + leaseSeconds - The period for which the subscription is expected to be active in the `hub`
-# + webHookConfig - HTTP client configurations for subscription/unsubscription intent verification
+# + leaseSeconds - The period for which the subscription is expected to be active in the `hub`  
+# + webHookConfig - HTTP client configurations for subscription/unsubscription intent verification  
+# + autoVerifySubscriptionIntent - Configuration to enable or disable automatic subscription intent verification
 public type ServiceConfiguration record {|
     int leaseSeconds?;
     ClientConfiguration webHookConfig?;
+    boolean autoVerifySubscriptionIntent = false;
 |};
 
 # WebSub Hub Configuration for the service.

ballerina/commons.bal

@@ -54,6 +54,9 @@ const int LISTENER_START_ERROR = -3;
 const int LISTENER_DETACH_ERROR = -4;
 const int LISTENER_STOP_ERROR = -5;
 const int CLIENT_INIT_ERROR = -10;
+const SUB_AUTO_VERIFY_ERROR = -11;
+
+const DEFAULT_HUB_LEASE_SECONDS = 86400;
 
 # Options to compress using Gzip or deflate.
 #

ballerina/http_service.bal

@@ -26,12 +26,13 @@ isolated service class HttpService {
     private final int defaultLeaseSeconds;
     private final SubscriptionHandler subscriptionHandler;
 
-    isolated function init(HttpToWebsubhubAdaptor adaptor, string hubUrl, int leaseSeconds,
-            *ClientConfiguration clientConfig) {
+    isolated function init(HttpToWebsubhubAdaptor adaptor, string hubUrl, ServiceConfiguration? serviceConfig) {
         self.adaptor = adaptor;
         self.hub = hubUrl;
-        self.defaultLeaseSeconds = leaseSeconds;
-        self.subscriptionHandler = new (adaptor, clientConfig);
+        self.defaultLeaseSeconds = serviceConfig?.leaseSeconds ?: DEFAULT_HUB_LEASE_SECONDS;
+        ClientConfiguration clientConfig = serviceConfig?.webHookConfig ?: {};
+        boolean autoVerifySubscriptionIntent = serviceConfig?.autoVerifySubscriptionIntent ?: false;
+        self.subscriptionHandler = new (adaptor, autoVerifySubscriptionIntent, clientConfig);
     }
 
     isolated resource function post .(http:Caller caller, http:Request request, http:Headers headers) returns Error? {

ballerina/hub_controller.bal

@@ -0,0 +1,58 @@
+// Copyright (c) 2025 WSO2 LLC. (http://www.wso2.com).
+//
+// WSO2 LLC. 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.
+
+# Component which can use to change the default subcription intent verification flow.
+public isolated class Controller {
+    private final boolean autoVerifySubscriptionIntent;
+
+    private final map<Subscription|Unsubscription> autoVerifyState = {};
+
+    isolated function init(boolean autoVerifySubscriptionIntent) {
+        self.autoVerifySubscriptionIntent = autoVerifySubscriptionIntent;
+    }
+
+    # Marks a particular subscription as verified.
+    # 
+    # + subscription - The `websubhub:Subscription` or `websubhub:Unsubscription` message
+    # + return - A `websubhub:Error` if the `websubhub:Service` has not enabled subscription auto-verification, 
+    #           or else nil
+    public isolated function markAsVerified(Subscription|Unsubscription subscription) returns Error? {
+        if !self.autoVerifySubscriptionIntent {
+            return error Error(
+                "Trying mark a subcription as verified, but the `hub` has not enabled automatic subscription intent verification", 
+                statusCode = SUB_AUTO_VERIFY_ERROR);
+        }
+
+        string 'key = constructSubscriptionKey(subscription);
+        lock {
+            self.autoVerifyState['key] = subscription.cloneReadOnly();
+        }
+    }
+
+    isolated function skipSubscriptionVerification(Subscription|Unsubscription subscription) returns boolean {
+        string 'key = constructSubscriptionKey(subscription);
+        Subscription|Unsubscription? skipped;
+        lock {
+            skipped = self.autoVerifyState.removeIfHasKey('key).cloneReadOnly();
+        }
+        return skipped !is ();
+    }
+}
+
+isolated function constructSubscriptionKey(record {} message) returns string {
+    string[] values = message.toArray().'map(v => string `${v.toString()}`);
+    return string:'join(":::", ...values);
+}

ballerina/hub_listener.bal

@@ -19,7 +19,6 @@ import ballerina/log;
 
 # Represents a Service listener endpoint.
 public class Listener {
-    private final int defaultHubLeaseSeconds = 864000;
     private http:Listener httpListener;
     private http:InferredListenerConfiguration listenerConfig;
     private int port;
@@ -65,18 +64,9 @@ public class Listener {
 
         string hubUrl = self.retrieveHubUrl(name);
         ServiceConfiguration? configuration = retrieveServiceAnnotations('service);
-        HttpToWebsubhubAdaptor adaptor = new('service);
-        if configuration is ServiceConfiguration {
-            int leaseSeconds = configuration?.leaseSeconds is int ? <int>(configuration?.leaseSeconds) : self.defaultHubLeaseSeconds;
-            if configuration?.webHookConfig is ClientConfiguration {
-                self.httpService = new(adaptor, hubUrl, leaseSeconds, <ClientConfiguration>(configuration?.webHookConfig));
-            } else {
-                self.httpService = new(adaptor, hubUrl, leaseSeconds);
-            }
-        } else {
-            self.httpService = new(adaptor, hubUrl, self.defaultHubLeaseSeconds);
-        }
-        error? result = self.httpListener.attach(<HttpService> self.httpService, name);
+        HttpToWebsubhubAdaptor adaptor = new ('service);
+        self.httpService = new (adaptor, hubUrl, configuration);
+        error? result = self.httpListener.attach(<HttpService>self.httpService, name);
         if (result is error) {
             return error Error("Error occurred while attaching the service", result, statusCode = LISTENER_ATTACH_ERROR);
         }
@@ -92,9 +82,7 @@ public class Listener {
     isolated function retrieveHubUrl(string[]|string? servicePath) returns string {
         string host = self.listenerConfig.host;
         string protocol = self.listenerConfig.secureSocket is () ? "http" : "https";
-
-        string concatenatedServicePath = "";
-
+        string concatenatedServicePath = "";        
         if servicePath is string {
             concatenatedServicePath += "/" + <string>servicePath;
         } else if servicePath is string[] {

ballerina/natives.bal

@@ -45,8 +45,9 @@ isolated class HttpToWebsubhubAdaptor {
         'class: "io.ballerina.stdlib.websubhub.NativeHttpToWebsubhubAdaptor"
     } external;
 
-    isolated function callOnSubscriptionMethod(Subscription msg, http:Headers headers) returns SubscriptionAccepted|
-        SubscriptionPermanentRedirect|SubscriptionTemporaryRedirect|BadSubscriptionError|InternalSubscriptionError|error = @java:Method {
+    isolated function callOnSubscriptionMethod(Subscription msg, http:Headers headers, Controller hubController) 
+    returns SubscriptionAccepted|SubscriptionPermanentRedirect|SubscriptionTemporaryRedirect|BadSubscriptionError
+    |InternalSubscriptionError|error = @java:Method {
         'class: "io.ballerina.stdlib.websubhub.NativeHttpToWebsubhubAdaptor"
     } external;
 
@@ -59,7 +60,7 @@ isolated class HttpToWebsubhubAdaptor {
         'class: "io.ballerina.stdlib.websubhub.NativeHttpToWebsubhubAdaptor"
     } external;
 
-    isolated function callOnUnsubscriptionMethod(Unsubscription msg, http:Headers headers)
+    isolated function callOnUnsubscriptionMethod(Unsubscription msg, http:Headers headers, Controller hubController) 
     returns UnsubscriptionAccepted|BadUnsubscriptionError|InternalUnsubscriptionError|error = @java:Method {
         'class: "io.ballerina.stdlib.websubhub.NativeHttpToWebsubhubAdaptor"
     } external;

ballerina/subscription.bal

@@ -19,15 +19,18 @@ import ballerina/uuid;
 
 isolated class SubscriptionHandler {
     private final HttpToWebsubhubAdaptor adaptor;
+    private final Controller hubController;
     private final readonly & ClientConfiguration clientConfig;
 
     private final boolean isOnSubscriptionAvailable;
     private final boolean isOnSubscriptionValidationAvailable;
     private final boolean isOnUnsubscriptionAvailable;
     private final boolean isOnUnsubscriptionValidationAvailable;
 
-    isolated function init(HttpToWebsubhubAdaptor adaptor, *ClientConfiguration clientConfig) {
+    isolated function init(HttpToWebsubhubAdaptor adaptor, boolean autoVerifySubscriptionIntent,
+            ClientConfiguration clientConfig) {
         self.adaptor = adaptor;
+        self.hubController = new (autoVerifySubscriptionIntent);
         self.clientConfig = clientConfig.cloneReadOnly();
         string[] methodNames = adaptor.getServiceMethodNames();
         self.isOnSubscriptionAvailable = methodNames.indexOf("onSubscription") is int;
@@ -43,7 +46,8 @@ isolated class SubscriptionHandler {
             return response;
         }
 
-        SubscriptionAccepted|Redirect|error result = self.adaptor.callOnSubscriptionMethod(message, headers);
+        SubscriptionAccepted|Redirect|error result = self.adaptor.callOnSubscriptionMethod(
+            message, headers, self.hubController);
         if result is Redirect {
             return result;
         }
@@ -63,17 +67,20 @@ isolated class SubscriptionHandler {
             return;
         }
 
-        string challenge = uuid:createType4AsString();
-        [string, string?][] params = [
-            [HUB_MODE, MODE_SUBSCRIBE],
-            [HUB_TOPIC, message.hubTopic],
-            [HUB_CHALLENGE, challenge],
-            [HUB_LEASE_SECONDS, message.hubLeaseSeconds]
-        ];
-        http:Response subscriberResponse = check sendNotification(message.hubCallback, params, self.clientConfig);
-        string responsePayload = check subscriberResponse.getTextPayload();
-        if challenge != responsePayload {
-            return;
+        boolean skipIntentVerification = self.hubController.skipSubscriptionVerification(message);
+        if !skipIntentVerification {
+            string challenge = uuid:createType4AsString();
+            [string, string?][] params = [
+                [HUB_MODE, MODE_SUBSCRIBE],
+                [HUB_TOPIC, message.hubTopic],
+                [HUB_CHALLENGE, challenge],
+                [HUB_LEASE_SECONDS, message.hubLeaseSeconds]
+            ];
+            http:Response subscriberResponse = check sendNotification(message.hubCallback, params, self.clientConfig);
+            string responsePayload = check subscriberResponse.getTextPayload();
+            if challenge != responsePayload {
+                return;
+            }
         }
 
         VerifiedSubscription verifiedSubscription = {
@@ -100,7 +107,8 @@ isolated class SubscriptionHandler {
             return response;
         }
 
-        UnsubscriptionAccepted|error result = self.adaptor.callOnUnsubscriptionMethod(message, headers);
+        UnsubscriptionAccepted|error result = self.adaptor.callOnUnsubscriptionMethod(
+            message, headers, self.hubController);
         return processOnUnsubscriptionResult(result);
     }
 
@@ -115,16 +123,19 @@ isolated class SubscriptionHandler {
             _ = check sendNotification(message.hubCallback, params, self.clientConfig);
         }
 
-        string challenge = uuid:createType4AsString();
-        [string, string?][] params = [
-            [HUB_MODE, MODE_UNSUBSCRIBE],
-            [HUB_TOPIC, message.hubTopic],
-            [HUB_CHALLENGE, challenge]
-        ];
-        http:Response subscriberResponse = check sendNotification(message.hubCallback, params, self.clientConfig);
-        string responsePayload = check subscriberResponse.getTextPayload();
-        if challenge != responsePayload {
-            return;
+        boolean skipIntentVerification = self.hubController.skipSubscriptionVerification(message);
+        if !skipIntentVerification {
+            string challenge = uuid:createType4AsString();
+            [string, string?][] params = [
+                [HUB_MODE, MODE_UNSUBSCRIBE],
+                [HUB_TOPIC, message.hubTopic],
+                [HUB_CHALLENGE, challenge]
+            ];
+            http:Response subscriberResponse = check sendNotification(message.hubCallback, params, self.clientConfig);
+            string responsePayload = check subscriberResponse.getTextPayload();
+            if challenge != responsePayload {
+                return;
+            }
         }
 
         VerifiedUnsubscription verifiedUnsubscription = {