Make the WARC digest algorithm configurable

docs/src/main/asciidoc/configuration.adoc

@@ -737,6 +737,7 @@ See the link:https://github.com/apache/stormcrawler/tree/main/external/warc[warc
 | key | default value | description
 
 | warc.metadata.keys | - | Metadata keys to include as WARC metadata records (optional list).
+| warc.digest.algorithm | sha1 | Algorithm for the WARC-Payload-Digest and WARC-Block-Digest headers: `sha1` or `sha256`. SHA-1 is the convention across the WARC ecosystem; downstream tooling such as CDX indexers may expect `sha1` digests. Digest values are Base32-encoded without padding.
 |===
 
 NOTE: For complete WARC records, set `http.store.headers` to `true`. The OkHttp protocol (`org.apache.stormcrawler.protocol.okhttp.HttpProtocol`) is recommended for WARC generation as it provides verbatim HTTP headers.

external/warc/README.md

@@ -161,6 +161,16 @@ A note on the recording of HTTP requests and responses with StormCrawler and the
 
 You can specify in the configuration which metadata key/values to store as WARC metadata using `warc.metadata.keys`.
 
+The algorithm used to compute the `WARC-Payload-Digest` and `WARC-Block-Digest` fields is configurable with `warc.digest.algorithm`. It accepts `sha1` (the default) and `sha256`:
+
+```
+  warc.digest.algorithm: sha256
+```
+
+The value is matched case-insensitively and an optional hyphen is ignored, i.e. `SHA-256` is also accepted. SHA-1 is the convention across the WARC ecosystem and downstream tooling such as CDX indexers may expect `sha1` digests, so change the default only if your downstream tooling supports the alternative algorithm. Invalid values make the bolt fail when it is prepared.
+
+The digest values are written as Base32 without the trailing `=` padding, as required by the grammar of the WARC digest fields (the digest value is a token, which does not allow the padding character).
+
 ## Consuming WARC files
 
 Web archives harvested in the [WARC format](https://iipc.github.io/warc-specifications/specifications/warc-format/warc-1.1/) can be used as input for StormCrawler – instead of fetching content from remote servers, the WARCSpout reads WARC files and emits the archive web page captures as tuples into the topology.

external/warc/src/main/java/org/apache/stormcrawler/warc/MetadataRecordFormat.java

@@ -51,8 +51,24 @@ public class MetadataRecordFormat extends WARCRecordFormat {
 
     private final List<String> metadataKeys;
 
+    /**
+     * Creates a metadata record format computing the digests with the default algorithm (SHA-1).
+     */
     public MetadataRecordFormat(List<String> metadataKeys) {
-        super("");
+        this(metadataKeys, WARCRecordFormat.DIGEST_ALGORITHM_SHA1);
+    }
+
+    /**
+     * Creates a metadata record format computing the WARC-Block-Digest field with the given
+     * algorithm.
+     *
+     * @param metadataKeys metadata keys to include as fields of the metadata record
+     * @param digestAlgorithm algorithm for the digest fields; see {@link
+     *     WARCRecordFormat#WARCRecordFormat(String, String)}
+     * @throws IllegalArgumentException if the value is not a supported algorithm
+     */
+    public MetadataRecordFormat(List<String> metadataKeys, String digestAlgorithm) {
+        super("", digestAlgorithm);
         // the keys are fixed configuration: validate them once here instead of
         // for every record
         final List<String> validKeys = new ArrayList<>(metadataKeys.size());
@@ -118,7 +134,7 @@ public byte[] format(Tuple tuple) {
         int contentLength = metadata_representation.length;
         buffer.append("Content-Length: ").append(Integer.toString(contentLength)).append(CRLF);
 
-        String blockDigest = getDigestSha1(metadata_representation);
+        String blockDigest = getDigest(metadata_representation);
 
         String captureTime = getCaptureTime(metadata);
         buffer.append("WARC-Date: ").append(captureTime).append(CRLF);

external/warc/src/main/java/org/apache/stormcrawler/warc/WARCHdfsBolt.java

@@ -74,14 +74,19 @@ public void doPrepare(
             throws IOException {
         super.doPrepare(conf, topologyContext, collector);
         protocolMDprefix = ConfUtils.getString(conf, ProtocolResponse.PROTOCOL_MD_PREFIX_PARAM, "");
-        withRecordFormat(new WARCRecordFormat(protocolMDprefix));
+        String digestAlgorithm =
+                ConfUtils.getString(
+                        conf,
+                        WARCRecordFormat.DIGEST_ALGORITHM_PARAM,
+                        WARCRecordFormat.DIGEST_ALGORITHM_SHA1);
+        withRecordFormat(new WARCRecordFormat(protocolMDprefix, digestAlgorithm));
         if (withRequestRecords) {
-            addRecordFormat(new WARCRequestRecordFormat(protocolMDprefix), 0);
+            addRecordFormat(new WARCRequestRecordFormat(protocolMDprefix, digestAlgorithm), 0);
         }
         // detect if a list of keys was specified to be stored in the metadata
         List<String> metadataToWrite = ConfUtils.loadListFromConf(METADATA_KEYS_STORE, conf);
         if (!metadataToWrite.isEmpty()) {
-            addRecordFormat(new MetadataRecordFormat(metadataToWrite), 1);
+            addRecordFormat(new MetadataRecordFormat(metadataToWrite, digestAlgorithm), 1);
         }
     }
 

external/warc/src/main/java/org/apache/stormcrawler/warc/WARCRecordFormat.java

@@ -33,6 +33,7 @@
 import java.util.Locale;
 import java.util.Map;
 import java.util.Map.Entry;
+import java.util.Objects;
 import java.util.Optional;
 import java.util.UUID;
 import java.util.regex.Pattern;
@@ -129,23 +130,132 @@ static String sanitizeWarcFieldValue(String value) {
             Pattern.compile("(?i)(?:Content-(?:Encoding|Length)|Transfer-Encoding)");
     protected static final String X_HIDE_HEADER = "X-Crawler-";
 
+    /**
+     * Configuration key setting the algorithm used to compute the WARC-Payload-Digest and
+     * WARC-Block-Digest fields. Supported values are {@value #DIGEST_ALGORITHM_SHA1} (the default)
+     * and {@value #DIGEST_ALGORITHM_SHA256}.
+     *
+     * <p>Note: SHA-1 is the convention across the WARC ecosystem and downstream tooling (CDX
+     * indexes, revisit record handling) may expect it. Change the default deliberately, not
+     * casually.
+     */
+    public static final String DIGEST_ALGORITHM_PARAM = "warc.digest.algorithm";
+
+    public static final String DIGEST_ALGORITHM_SHA1 = "sha1";
+
+    public static final String DIGEST_ALGORITHM_SHA256 = "sha256";
+
     private static final Base32 base32 = new Base32();
-    private static final String digestNoContent = getDigestSha1(new byte[0]);
 
     protected final String protocolMDprefix;
 
+    /** JCA name of the message digest algorithm, e.g. &quot;SHA-1&quot;. */
+    private final String digestJCAName;
+
+    /** Algorithm prefix of the WARC digest fields, e.g. &quot;sha1:&quot;. */
+    private final String digestPrefix;
+
+    private final String digestNoContent;
+
+    /**
+     * Creates a record format computing the digests with the default algorithm (SHA-1).
+     *
+     * @param protocolMDprefix prefix of the metadata keys holding the protocol response, as set by
+     *     {@code protocol.md.prefix}; may be empty
+     */
     public WARCRecordFormat(String protocolMDprefix) {
+        this(protocolMDprefix, DIGEST_ALGORITHM_SHA1);
+    }
+
+    /**
+     * Creates a record format computing the WARC-Payload-Digest and WARC-Block-Digest fields with
+     * the given algorithm.
+     *
+     * @param protocolMDprefix prefix of the metadata keys holding the protocol response, as set by
+     *     {@code protocol.md.prefix}; may be empty
+     * @param digestAlgorithm algorithm for the digest fields, {@value #DIGEST_ALGORITHM_SHA1} (the
+     *     default) or {@value #DIGEST_ALGORITHM_SHA256}; matched case-insensitively with an
+     *     optional hyphen, surrounding whitespace is trimmed. A {@code null} or blank value selects
+     *     the default SHA-1.
+     * @throws IllegalArgumentException if the value is not a supported algorithm
+     */
+    public WARCRecordFormat(String protocolMDprefix, String digestAlgorithm) {
         this.protocolMDprefix = protocolMDprefix;
+        this.digestJCAName = getDigestJCAName(digestAlgorithm);
+        this.digestPrefix = digestJCAName.toLowerCase(Locale.ROOT).replace("-", "") + ":";
+        this.digestNoContent = getDigest(new byte[0]);
     }
 
-    public static String getDigestSha1(byte[] bytes) {
-        return "sha1:" + base32.encodeAsString(DigestUtils.sha1(bytes));
+    /**
+     * Resolve the configured digest algorithm to the JCA name of the message digest. The value is
+     * matched case-insensitively and an optional hyphen is ignored, i.e. &quot;sha256&quot;,
+     * &quot;SHA-256&quot; etc. are all accepted.
+     *
+     * @param digestAlgorithm algorithm to resolve; {@code null} or blank selects the default SHA-1
+     * @return the JCA name of the message digest, e.g. &quot;SHA-1&quot;
+     * @throws IllegalArgumentException if the value is not a supported algorithm
+     */
+    private static String getDigestJCAName(String digestAlgorithm) {
+        if (StringUtils.isBlank(digestAlgorithm)) {
+            return "SHA-1";
+        }
+        return switch (digestAlgorithm.trim().toLowerCase(Locale.ROOT).replace("-", "")) {
+            case DIGEST_ALGORITHM_SHA1 -> "SHA-1";
+            case DIGEST_ALGORITHM_SHA256 -> "SHA-256";
+            default ->
+                    throw new IllegalArgumentException(
+                            "Unsupported value ["
+                                    + digestAlgorithm
+                                    + "] for "
+                                    + DIGEST_ALGORITHM_PARAM
+                                    + ", supported algorithms: "
+                                    + DIGEST_ALGORITHM_SHA1
+                                    + ", "
+                                    + DIGEST_ALGORITHM_SHA256);
+        };
+    }
+
+    /**
+     * Compute the digest of the given bytes with the configured algorithm.
+     *
+     * @param bytes bytes to digest, must not be null
+     * @return digest in the form &quot;&lt;algorithm&gt;:&lt;base32&gt;&quot;, e.g.
+     *     &quot;sha1:...&quot;
+     * @throws NullPointerException if the input is null
+     */
+    public String getDigest(byte[] bytes) {
+        Objects.requireNonNull(bytes, "bytes to digest must not be null");
+        MessageDigest md = DigestUtils.getDigest(digestJCAName);
+        return digestPrefix + base32Unpadded(md.digest(bytes));
+    }
+
+    /**
+     * Compute the digest of the concatenation of the two given byte arrays with the configured
+     * algorithm.
+     *
+     * @param bytes1 first bytes to digest, must not be null
+     * @param bytes2 second bytes to digest, must not be null
+     * @return digest in the form &quot;&lt;algorithm&gt;:&lt;base32&gt;&quot;, e.g.
+     *     &quot;sha1:...&quot;
+     * @throws NullPointerException if one of the inputs is null
+     */
+    public String getDigest(byte[] bytes1, byte[] bytes2) {
+        Objects.requireNonNull(bytes1, "first bytes to digest must not be null");
+        Objects.requireNonNull(bytes2, "second bytes to digest must not be null");
+        MessageDigest md = DigestUtils.getDigest(digestJCAName);
+        md.update(bytes1);
+        return digestPrefix + base32Unpadded(md.digest(bytes2));
     }
 
-    public static String getDigestSha1(byte[] bytes1, byte[] bytes2) {
-        MessageDigest sha1 = DigestUtils.getSha1Digest();
-        sha1.update(bytes1);
-        return "sha1:" + base32.encodeAsString(sha1.digest(bytes2));
+    /**
+     * Base32-encode a digest value without the trailing &quot;=&quot; padding characters: the WARC
+     * digest fields define the digest value as a token, which does not allow the padding character
+     * (cf. ISO 28500 WARC 1.1, WARC-Block-Digest / WARC-Payload-Digest). SHA-1 digests are
+     * unaffected (32 characters without padding), while e.g. SHA-256 digests would end in
+     * &quot;====&quot;.
+     */
+    private static String base32Unpadded(byte[] digest) {
+        return StringUtils.stripEnd(base32.encodeAsString(digest), "=");
     }
 
     /** Generates a WARC info entry which can be stored at the beginning of each WARC file. */
@@ -435,14 +545,14 @@ public byte[] format(Tuple tuple) {
         String blockDigest = digestNoContent;
         if (content != null) {
             contentLength = content.length;
-            payloadDigest = getDigestSha1(content);
+            payloadDigest = getDigest(content);
             if (WARCTypeValue.equals(WARC_TYPE_RESPONSE)) {
-                blockDigest = getDigestSha1(httpheaders, content);
+                blockDigest = getDigest(httpheaders, content);
             } else {
                 blockDigest = payloadDigest;
             }
         } else if (WARCTypeValue.equals(WARC_TYPE_RESPONSE)) {
-            blockDigest = getDigestSha1(httpheaders);
+            blockDigest = getDigest(httpheaders);
         }
 
         // add the length of the http header

external/warc/src/main/java/org/apache/stormcrawler/warc/WARCRequestRecordFormat.java

@@ -41,10 +41,25 @@ public class WARCRequestRecordFormat extends WARCRecordFormat {
     protected static final Pattern REQUEST_LINE_PATTERN =
             Pattern.compile("^\\S+ \\S+ HTTP/1\\.[01]$");
 
+    /** Creates a request record format computing the digests with the default algorithm (SHA-1). */
     public WARCRequestRecordFormat(String protocolMDprefix) {
         super(protocolMDprefix);
     }
 
+    /**
+     * Creates a request record format computing the WARC-Block-Digest field with the given
+     * algorithm.
+     *
+     * @param protocolMDprefix prefix of the metadata keys holding the protocol response, as set by
+     *     {@code protocol.md.prefix}; may be empty
+     * @param digestAlgorithm algorithm for the digest fields; see {@link
+     *     WARCRecordFormat#WARCRecordFormat(String, String)}
+     * @throws IllegalArgumentException if the value is not a supported algorithm
+     */
+    public WARCRequestRecordFormat(String protocolMDprefix, String digestAlgorithm) {
+        super(protocolMDprefix, digestAlgorithm);
+    }
+
     @Override
     public byte[] format(Tuple tuple) {
 
@@ -82,7 +97,7 @@ public byte[] format(Tuple tuple) {
         int contentLength = httpheaders.length;
         buffer.append("Content-Length: ").append(Integer.toString(contentLength)).append(CRLF);
 
-        String blockDigest = getDigestSha1(httpheaders);
+        String blockDigest = getDigest(httpheaders);
 
         String captureTime = getCaptureTime(metadata);
         buffer.append("WARC-Date: ").append(captureTime).append(CRLF);