Add chunked signature support for streaming uploads

src/ChunkedSignatureSigner.php

@@ -0,0 +1,94 @@
+<?php
+
+namespace Litebase;
+
+/**
+ * ChunkedSignatureSigner handles signature calculation for chunked uploads in LQTP.
+ * This implements chunked signature validation similar to AWS Signature Version 4.
+ */
+class ChunkedSignatureSigner
+{
+    protected ?string $accessKeySecret;
+
+    protected string $date;
+
+    protected string $previousSignature;
+
+    /**
+     * Create a new ChunkedSignatureSigner instance.
+     */
+    public function __construct(?string $accessKeySecret, string $date, string $seedSignature)
+    {
+        $this->accessKeySecret = $accessKeySecret;
+        $this->date = $date;
+        $this->previousSignature = $seedSignature;
+    }
+
+    /**
+     * Sign a chunk of data and return the signature.
+     *
+     * Signature calculation (LQTP protocol):
+     *  1. Hash the chunk data: chunkHash = SHA256(chunkData)
+     *  2. Create string to sign: stringToSign = previousSignature + chunkHash
+     *  3. Generate signing key chain:
+     *     - dateKey = HMAC-SHA256(accessKeySecret, date)
+     *     - serviceKey = HMAC-SHA256(dateKey, "litebase_request")
+     *  4. Sign: signature = HMAC-SHA256(serviceKey, stringToSign)
+     *
+     * The signature chains ensure chunks are sent in the correct order and prevents tampering.
+     */
+    public function signChunk(string $chunkData): string
+    {
+        // Calculate the hash of the chunk data
+        $chunkHash = hash('sha256', $chunkData);
+
+        // Create the string to sign for this chunk
+        // Format: previousSignature + chunkHash
+        $stringToSign = $this->previousSignature . $chunkHash;
+
+        // Create the signing key chain (same as in request signature validation)
+        $dateKey = hash_hmac('sha256', $this->date, $this->accessKeySecret ?? '');
+        $serviceKey = hash_hmac('sha256', 'litebase_request', $dateKey);
+
+        // Sign the chunk
+        $signature = hash_hmac('sha256', $stringToSign, $serviceKey);
+
+        // Update the previous signature for the next chunk
+        $this->previousSignature = $signature;
+
+        return $signature;
+    }
+
+    /**
+     * Get the current previous signature (for testing/debugging).
+     */
+    public function getPreviousSignature(): string
+    {
+        return $this->previousSignature;
+    }
+
+    /**
+     * Extract the signature from a base64 encoded authorization token.
+     *
+     * @param  string  $token  The base64 encoded token
+     * @return string|null The extracted signature or null if not found
+     */
+    public static function extractSignatureFromToken(string $token): ?string
+    {
+        $decoded = base64_decode($token, true);
+
+        if ($decoded === false) {
+            return null;
+        }
+
+        $parts = explode(';', $decoded);
+
+        foreach ($parts as $part) {
+            if (str_starts_with($part, 'signature=')) {
+                return substr($part, strlen('signature='));
+            }
+        }
+
+        return null;
+    }
+}

src/Configuration.php

@@ -9,9 +9,9 @@
  */
 class Configuration extends BaseConfiguration
 {
-    protected string $accessKeyId = '';
+    protected ?string $accessKeyId;
 
-    protected string $accessKeySecret = '';
+    protected ?string $accessKeySecret;
 
     protected ?string $database = '';
 
@@ -24,15 +24,15 @@ class Configuration extends BaseConfiguration
      */
     public function getAccessKeyId(): string
     {
-        return $this->accessKeyId;
+        return $this->accessKeyId ?? '';
     }
 
     /**
      * Get the access key secret
      */
     public function getAccessKeySecret(): string
     {
-        return $this->accessKeySecret;
+        return $this->accessKeySecret ?? '';
     }
 
     /**
@@ -70,7 +70,7 @@ public function getPort(): ?string
     /**
      * Set access key credentials for HMAC-SHA256 authentication
      */
-    public function setAccessKey(string $accessKeyId, string $accessKeySecret): self
+    public function setAccessKey(?string $accessKeyId, ?string $accessKeySecret): self
     {
         $this->accessKeyId = $accessKeyId;
         $this->accessKeySecret = $accessKeySecret;

src/Connection.php

@@ -42,23 +42,26 @@ class Connection
     /** @var \Fiber<never, void, \Throwable|null, \Throwable|null>|null */
     protected ?Fiber $writer;
 
+    protected ?ChunkedSignatureSigner $chunkedSigner;
+
     /**
      * Create a new connection instance.
      *
      * @param  array<string, int|string>  $requestHeaders
      */
-    public function __construct(public string $url, public array $requestHeaders = [])
+    public function __construct(public string $url, public array $requestHeaders = [], ?ChunkedSignatureSigner $chunkedSigner = null)
     {
-        $host = parse_url($url, PHP_URL_HOST);
+        $host = parse_url($this->url, PHP_URL_HOST);
 
         if ($host === false || $host === null) {
             throw new Exception('[Litebase Client Error]: Invalid URL provided');
         }
+
         $this->host = $host;
 
-        $this->port = parse_url($url, PHP_URL_PORT) ?: 80;
+        $this->port = parse_url($this->url, PHP_URL_PORT) ?: 80;
 
-        $path = parse_url($url, PHP_URL_PATH);
+        $path = parse_url($this->url, PHP_URL_PATH);
 
         if ($path === false || $path === null) {
             throw new Exception('[Litebase Client Error]: Invalid URL provided');
@@ -71,6 +74,7 @@ public function __construct(public string $url, public array $requestHeaders = [
         }
 
         $this->queryRequestEncoder = new QueryRequestEncoder;
+        $this->chunkedSigner = $chunkedSigner;
     }
 
     /**
@@ -357,7 +361,7 @@ public function open(): void
         stream_set_timeout($this->socket, 5);
 
         $error = fwrite($this->socket, "POST {$this->path} HTTP/1.1\r\n");
-        $error = fwrite($this->socket, implode("\r\n", $this->headers)."\r\n");
+        $error = fwrite($this->socket, implode("\r\n", $this->headers) . "\r\n");
         $error = fwrite($this->socket, "\r\n");
 
         if ($error === false) {
@@ -367,7 +371,7 @@ public function open(): void
         $this->open = true;
 
         $this->messages = [
-            pack('C', QueryStreamMessageType::OPEN_CONNECTION->value.pack('V', 0)),
+            pack('C', QueryStreamMessageType::OPEN_CONNECTION->value . pack('V', 0)),
             ...$this->messages,
         ];
 
@@ -381,7 +385,28 @@ public function send(Query $query): QueryResult
     {
         $queryRequest = $this->queryRequestEncoder->encode($query);
 
-        $frame = pack('C', QueryStreamMessageType::FRAME->value).pack('V', strlen($queryRequest)).$queryRequest;
+        // If chunked signer is available, create a signed frame per LQTP protocol
+        if ($this->chunkedSigner !== null) {
+            // Frame data: [QueryLength:4][QueryData]
+            $frameData = pack('V', strlen($queryRequest)) . $queryRequest;
+
+            // Sign the frame data using chunked signature scheme (similar to AWS Sig4)
+            $chunkSignature = $this->chunkedSigner->signChunk($frameData);
+
+            // Build complete frame with signature per LQTP protocol
+            // Frame format: [MessageType:1][FrameLength:4][SignatureLength:4][Signature:N][FrameData]
+            $signatureBytes = $chunkSignature;
+            $totalLength = 4 + strlen($signatureBytes) + strlen($frameData);
+
+            $frame = pack('C', QueryStreamMessageType::FRAME->value) // Message type (0x04)
+                . pack('V', $totalLength)                            // Total length (signature metadata + frame data)
+                . pack('V', strlen($signatureBytes))                 // Signature length
+                . $signatureBytes                                     // Hex-encoded chunk signature
+                . $frameData;                                         // Frame data
+        } else {
+            // Fallback to unsigned frame format (deprecated)
+            $frame = pack('C', QueryStreamMessageType::FRAME->value) . pack('V', strlen($queryRequest)) . $queryRequest;
+        }
 
         $this->messages[] = $frame;
 
@@ -410,7 +435,7 @@ public function send(Query $query): QueryResult
 
                             continue;
                         } catch (Exception $reconnectException) {
-                            throw new Exception('[Litebase Client Error]: Failed to reconnect after connection loss: '.$reconnectException->getMessage());
+                            throw new Exception('[Litebase Client Error]: Failed to reconnect after connection loss: ' . $reconnectException->getMessage());
                         }
                     }
                 }
@@ -565,7 +590,7 @@ protected function writeMessage(string $message): void
         $chunkSize = dechex(strlen($message));
 
         $n = $this->socket ?
-            fwrite($this->socket, $chunkSize."\r\n".$message."\r\n") :
+            fwrite($this->socket, $chunkSize . "\r\n" . $message . "\r\n") :
             false;
 
         if ($n === false) {

src/HttpStreamingTransport.php

@@ -12,6 +12,8 @@ class HttpStreamingTransport implements TransportInterface
 
     protected Connection $connection;
 
+    protected ?ChunkedSignatureSigner $chunkedSigner = null;
+
     /**
      * Create a new instance of the transport.
      */
@@ -21,6 +23,12 @@ public function __construct(
 
     public function send(Query $query): ?QueryResult
     {
+        if (empty($this->config->getDatabase()) || empty($this->config->getBranch())) {
+            throw new LitebaseConnectionException(
+                message: '[Litebase Client Error] Database and Branch names must be set for the streaming transport',
+            );
+        }
+
         $path = sprintf(
             'v1/databases/%s/branches/%s/query/stream',
             $this->config->getDatabase(),
@@ -41,28 +49,37 @@ public function send(Query $query): ?QueryResult
                 ? sprintf('https://%s/%s', $this->config->getHost(), $path)
                 : sprintf('http://%s:%d/%s', $this->config->getHost(), $this->config->getPort(), $path);
 
-            if (! empty($this->config->getUsername()) || ! (empty($this->config->getPassword()))) {
-                $headers['Authorization'] = 'Basic '.base64_encode($this->config->getUsername().':'.$this->config->getPassword());
+            if (empty($this->config->getAccessKeyId())) {
+                throw new LitebaseConnectionException(
+                    message: '[Litebase Client Error] An Access key is required for the streaming transport',
+                );
             }
 
-            if (! empty($this->config->getAccessToken())) {
-                $headers['Authorization'] = 'Bearer '.$this->config->getAccessToken();
-            }
+            // Use the streaming payload marker for chunked signature validation
+            $token = $this->getToken(
+                accessKeyID: $this->config->getAccessKeyId(),
+                accessKeySecret: $this->config->getAccessKeySecret(),
+                method: 'POST',
+                path: $path,
+                headers: $headers,
+                data: 'STREAMING-LITEBASE-HMAC-SHA256-PAYLOAD',
+            );
 
-            if (! empty($this->config->getAccessKeyId())) {
-                $token = $this->getToken(
-                    accessKeyID: $this->config->getAccessKeyId(),
-                    accessKeySecret: $this->config->getAccessKeySecret(),
-                    method: 'POST',
-                    path: $path,
-                    headers: $headers,
-                    data: null,
-                );
+            $headers['Authorization'] = sprintf('Litebase-HMAC-SHA256 %s', $token);
 
-                $headers['Authorization'] = sprintf('Litebase-HMAC-SHA256 %s', $token);
+            // Extract the seed signature from the token for chunk signing
+            $seedSignature = ChunkedSignatureSigner::extractSignatureFromToken($token);
+
+            if ($seedSignature !== null) {
+                // Create the chunked signature signer with the seed signature
+                $this->chunkedSigner = new ChunkedSignatureSigner(
+                    $this->config->getAccessKeySecret(),
+                    $headers['X-Litebase-Date'],
+                    $seedSignature
+                );
             }
 
-            $this->connection = new Connection($url, $headers);
+            $this->connection = new Connection($url, $headers, $this->chunkedSigner);
         }
 
         try {

src/LitebaseClient.php

@@ -220,7 +220,7 @@ public function withTransport(string $transportType): LitebaseClient
                 $this->transport = new HttpStreamingTransport($this->configuration);
                 break;
             default:
-                throw new Exception('Invalid transport type: '.$transportType);
+                throw new Exception('Invalid transport type: ' . $transportType);
         }
 
         return $this;

src/Middleware/AuthMiddleware.php

@@ -46,16 +46,13 @@ private function signRequest(RequestInterface $request): RequestInterface
             contentLength: strlen($body)
         );
 
-        $decodedBody = json_decode($body, true);
-        $data = is_array($decodedBody) ? $decodedBody : null;
-
         $token = $this->getToken(
             accessKeyID: $this->config->getAccessKeyId(),
             accessKeySecret: $this->config->getAccessKeySecret(),
             method: $request->getMethod(),
             path: $request->getUri()->getPath(),
             headers: $headers,
-            data: $data,
+            data: $body,
         );
 
         // Add signed headers to request

src/QueryRequestEncoder.php

@@ -9,7 +9,7 @@ public static function encode(Query $query): string
         $binaryData = '';
         $id = $query->id;
         $idLength = pack('V', strlen($id));
-        $binaryData .= $idLength.$id;
+        $binaryData .= $idLength . $id;
 
         $transactionIdLength = pack('V', strlen($query->transactionId ?? ''));
         $binaryData .= $transactionIdLength;
@@ -20,7 +20,7 @@ public static function encode(Query $query): string
 
         $statement = $query->statement;
         $statementLength = pack('V', strlen($statement));
-        $binaryData .= $statementLength.$statement;
+        $binaryData .= $statementLength . $statement;
 
         $parametersBinary = '';
 
@@ -72,15 +72,14 @@ public static function encode(Query $query): string
             $parameterType = pack('C', $parameterType);
 
             // Parameter value with length prefix (4 bytes little-endian + value)
-            $parameterValueWithLength = pack('V', $parameterValueLength).$parameterValue;
+            $parameterValueWithLength = pack('V', $parameterValueLength) . $parameterValue;
 
-            $parametersBinary .= $parameterType.$parameterValueWithLength;
+            $parametersBinary .= $parameterType . $parameterValueWithLength;
         }
 
         $parametersBinaryLength = pack('V', strlen($parametersBinary));
-        $binaryData .= $parametersBinaryLength.$parametersBinary;
-        $queryBinary = pack('V', strlen($binaryData)).$binaryData;
+        $binaryData .= $parametersBinaryLength . $parametersBinary;
 
-        return $queryBinary;
+        return $binaryData;
     }
 }