diff --git a/maws/README.md b/maws/README.md index 449bfea..f415f0a 100644 --- a/maws/README.md +++ b/maws/README.md @@ -90,6 +90,7 @@ That's it. `maws` wires everything to MAS, handles webhook GET/POST, background | `TELEGRAM_WEBHOOK_SECRET_TOKEN` | No | none | Provider-supported Telegram webhook secret token. | | `WHATSAPP_VERIFY_SIGNATURE` | No | `"false"` | When `true`, verify WhatsApp `X-Hub-Signature-256`. | | `WHATSAPP_APP_SECRET` | If verifying | none | Meta app secret used for WhatsApp signature verification. | +| `MAWS_HISTORY_WRITE_SECRET` | For history writes | none | Shared secret used to authenticate external history append requests. | | `MAWS_MANAGER_KWARGS_JSON` | No | `"{}"` | Extra kwargs for `AgentSystemManager`; MAWS-managed paths still win. | | `MAWS_BOT_KWARGS_JSON` | No | `"{}"` | Extra kwargs for the selected MAS bot constructor. | | `CAPTURE_FAILED_EVENTS` | No | `"false"` | Write failed jobs/events to S3 under `failed-events/`. | @@ -237,8 +238,38 @@ MAWS only uses provider-supported webhook security mechanisms: * **Telegram**: set `TELEGRAM_WEBHOOK_SECRET_TOKEN` in `.env.prod`. During deploy, MAWS passes it to Telegram `setWebhook` as `secret_token`. At runtime, MAWS requires incoming POSTs to include the matching `X-Telegram-Bot-Api-Secret-Token` header. * **WhatsApp**: keep the existing GET verify-token flow. For POST authenticity, set `webhook_security.whatsapp_verify_signature=true` in `params.json` and set `WHATSAPP_APP_SECRET` in `.env.prod`. MAWS verifies `X-Hub-Signature-256` against the raw request body. +The `/history/outgoing` endpoint is separate from the provider webhook and always requires `MAWS_HISTORY_WRITE_SECRET`. It verifies `X-MAWS-Timestamp` and `X-MAWS-Signature`; the signature is `sha256=` plus the HMAC-SHA256 of `.`. Requests more than five minutes from the current time are rejected. + Both checks are opt-in so existing bots do not break. Production deployments should enable the matching provider mechanism. +### Appending externally sent messages + +POST to `/history/outgoing` with a JSON body containing the recipient's MAWS `chat_id`, a stable `event_id`, and the message text. MAWS queues a worker job that appends the message with the `assistant` role, under the normal per-user lock and S3 history import/export lifecycle. It does not call `process_webhook_update`. The event ID is stored in block metadata so retries do not append the same message twice. + +```json +{ + "chat_id": "5491100000000", + "event_id": "wamid.outbound-1", + "content": "Hola desde el envio masivo" +} +``` + +Sign the exact serialized request body and send both headers: + +```js +const timestamp = String(Math.floor(Date.now() / 1000)); +const signature = createHmac("sha256", historyWriteSecret) + .update(`${timestamp}.${rawBody}`) + .digest("hex"); + +headers: { + "X-MAWS-Timestamp": timestamp, + "X-MAWS-Signature": `sha256=${signature}`, +} +``` + +Store the same random secret in MAWS `.env.prod` (so it reaches Lambda through SSM) and in the sending service. A `202` response means the history write was accepted for processing, not that the WhatsApp message was delivered. + --- ## Loading environment from SSM diff --git a/maws/resources/bootstrap.sh b/maws/resources/bootstrap.sh index e40b5c9..3d9d919 100644 --- a/maws/resources/bootstrap.sh +++ b/maws/resources/bootstrap.sh @@ -430,6 +430,12 @@ else EOF fi +cat >> template.yaml <<'EOF' + HistoryAppendPOST: + Type: Api + Properties: { Path: "/history/outgoing", Method: post } +EOF + if [[ "${BUSY_POLICY}" == "fifo" ]]; then cat >> template.yaml < Optional[dict]: + secret = getattr(self, "history_write_secret", "") + if not secret: + return { + "statusCode": 503, + "body": json.dumps("MAWS_HISTORY_WRITE_SECRET is not configured."), + } + + headers = event.get("headers") or {} + timestamp = _header_value(headers, "x-maws-timestamp") + signature = _header_value(headers, "x-maws-signature") + if ( + not timestamp.isascii() + or not timestamp.isdigit() + or not signature.isascii() + ): + return {"statusCode": 403, "body": json.dumps("Forbidden")} + try: + timestamp_seconds = int(timestamp) + except (TypeError, ValueError): + return {"statusCode": 403, "body": json.dumps("Forbidden")} + + if abs(time.time() - timestamp_seconds) > 300: + return {"statusCode": 403, "body": json.dumps("Forbidden")} + + signed_body = timestamp.encode("ascii") + b"." + raw_body + expected = "sha256=" + hmac.new( + secret.encode("utf-8"), signed_body, hashlib.sha256 + ).hexdigest() + if not hmac.compare_digest(signature, expected): + return {"statusCode": 403, "body": json.dumps("Forbidden")} + return None + + def _handle_history_append(self, event: dict, context): + raw_body = _event_body_bytes(event) + security_response = self._verify_history_write_signature(event, raw_body) + if security_response is not None: + return security_response + + try: + data = json.loads(raw_body.decode("utf-8") or "{}") + except (UnicodeDecodeError, json.JSONDecodeError): + return {"statusCode": 400, "body": json.dumps("Invalid JSON body.")} + if not isinstance(data, dict): + return {"statusCode": 400, "body": json.dumps("JSON body must be an object.")} + + chat_id = data.get("chat_id") + event_id = data.get("event_id") + content = data.get("content") + if ( + not isinstance(chat_id, str) + or not chat_id.strip() + or len(chat_id) > 256 + or not isinstance(event_id, str) + or not event_id.strip() + or len(event_id) > 256 + or not isinstance(content, str) + or not content.strip() + or len(content) > 65536 + ): + return { + "statusCode": 400, + "body": json.dumps("chat_id, event_id and content are required."), + } + + job = { + "maws_job_version": 1, + "provider": "history", + "bot_type": getattr(self, "bot_type", "whatsapp"), + "chat_id": chat_id.strip(), + "event_id": event_id.strip(), + "operation": "append_history", + "payload": {"content": content}, + } + + if getattr(self, "busy_policy", "drop") == "fifo": + queued = self._enqueue_jobs([job]) + if queued["statusCode"] >= 300: + return queued + else: + try: + invocation = self.lambda_client.invoke( + FunctionName=context.invoked_function_arn, + InvocationType="Event", + Payload=json.dumps({"maws_job": job}), + ) + if invocation.get("StatusCode") != 202: + return {"statusCode": 502, "body": json.dumps("History append could not be queued.")} + except Exception as e: + self._capture_failed_event(job, "self_invoke_failed", e) + return {"statusCode": 502, "body": json.dumps("History append could not be queued.")} + + return { + "statusCode": 202, + "body": json.dumps({"status": "accepted", "event_id": event_id.strip()}), + } + + def _history_contains_external_event(self, chat_id: str, event_id: str) -> bool: + for message in self.manager.get_messages(chat_id): + blocks = message.get("message", []) + if isinstance(blocks, dict): + blocks = [blocks] + if not isinstance(blocks, list): + continue + for block in blocks: + if not isinstance(block, dict): + continue + metadata = block.get("metadata") + if isinstance(metadata, dict) and metadata.get("maws_event_id") == event_id: + return True + return False + def _handle_http_post(self, event: dict, context): print("[maws] Received webhook POST.") try: @@ -956,9 +1072,22 @@ def _process_job(self, job: dict, context=None, raise_on_error: bool = False): self._sync_user_files_from_s3(chat_id) self.manager.set_current_user(chat_id) - # Process - loop = self.get_event_loop() - loop.run_until_complete(self.bot_instance.process_webhook_update(job.get("payload") or {})) + if job.get("operation") == "append_history": + event_id = str(job.get("event_id") or "") + if not self._history_contains_external_event(chat_id, event_id): + self.manager.add_blocks( + [{ + "type": "text", + "content": {"response": job.get("payload", {}).get("content", "")}, + "metadata": {"maws_event_id": event_id}, + }], + role="assistant", + msg_type="assistant", + user_id=chat_id, + ) + else: + loop = self.get_event_loop() + loop.run_until_complete(self.bot_instance.process_webhook_update(job.get("payload") or {})) except Exception as e: print(f"[maws][process][ERR] {e}") traceback.print_exc() diff --git a/tests/test_maws_runtime.py b/tests/test_maws_runtime.py index 47650d0..8c1f811 100644 --- a/tests/test_maws_runtime.py +++ b/tests/test_maws_runtime.py @@ -3,6 +3,7 @@ import json import hmac import hashlib +import time import types import pytest @@ -201,6 +202,106 @@ def test_whatsapp_bad_signature_is_rejected_before_invocation(): assert response["statusCode"] == 403 +def test_history_append_requires_valid_hmac_and_queues_a_worker_job(): + runtime = object.__new__(MawsRuntime) + runtime.bot_type = "whatsapp" + runtime.history_write_secret = "history-secret" + runtime.busy_policy = "drop" + invocations = [] + runtime.lambda_client = types.SimpleNamespace( + invoke=lambda **kwargs: invocations.append(kwargs) or {"StatusCode": 202} + ) + body = json.dumps({ + "chat_id": "5491100000000", + "event_id": "wamid.outbound-1", + "content": "Hola desde el bulk", + }, separators=(",", ":")) + timestamp = str(int(time.time())) + signature = "sha256=" + hmac.new( + b"history-secret", + timestamp.encode("ascii") + b"." + body.encode("utf-8"), + hashlib.sha256, + ).hexdigest() + + response = MawsRuntime.handle_apigw_event( + runtime, + { + "requestContext": {"http": {"method": "POST"}}, + "rawPath": "/Prod/history/outgoing", + "headers": { + "X-MAWS-Timestamp": timestamp, + "X-MAWS-Signature": signature, + }, + "body": body, + }, + _Context(), + ) + + assert response["statusCode"] == 202 + assert len(invocations) == 1 + job = json.loads(invocations[0]["Payload"])["maws_job"] + assert job["operation"] == "append_history" + assert job["chat_id"] == "5491100000000" + assert job["event_id"] == "wamid.outbound-1" + assert job["payload"]["content"] == "Hola desde el bulk" + + +def test_history_append_rejects_invalid_signature(): + runtime = object.__new__(MawsRuntime) + runtime.bot_type = "whatsapp" + runtime.history_write_secret = "history-secret" + runtime.lambda_client = types.SimpleNamespace( + invoke=lambda **kwargs: (_ for _ in ()).throw(AssertionError("should not invoke")) + ) + body = json.dumps({ + "chat_id": "user-1", + "event_id": "evt-1", + "content": "sent", + }) + + response = MawsRuntime.handle_apigw_event( + runtime, + { + "requestContext": {"http": {"method": "POST"}}, + "rawPath": "/Prod/history/outgoing", + "headers": { + "X-MAWS-Timestamp": str(int(time.time())), + "X-MAWS-Signature": "sha256=bad", + }, + "body": body, + }, + _Context(), + ) + + assert response["statusCode"] == 403 + + +def test_history_append_requires_configured_secret(): + runtime = object.__new__(MawsRuntime) + body = json.dumps({ + "chat_id": "user-1", + "event_id": "evt-1", + "content": "sent", + }) + timestamp = str(int(time.time())) + + response = MawsRuntime.handle_apigw_event( + runtime, + { + "requestContext": {"http": {"method": "POST"}}, + "rawPath": "/Prod/history/outgoing", + "headers": { + "X-MAWS-Timestamp": timestamp, + "X-MAWS-Signature": "sha256=unused", + }, + "body": body, + }, + _Context(), + ) + + assert response["statusCode"] == 503 + + @pytest.mark.parametrize( ("event", "expected"), [ @@ -336,6 +437,87 @@ async def process_webhook_update(self, update): assert runtime.s3.puts[("history-bucket", "history/123.sqlite")] == b"new-history" +def test_worker_appends_outgoing_history_without_processing_webhook(workspace_tmp_path): + runtime = object.__new__(MawsRuntime) + runtime.bot_type = "whatsapp" + runtime.bucket_name = "history-bucket" + runtime.history_prefix = "history" + runtime.history_mode = "per_user" + runtime.TMP_DIR = str(workspace_tmp_path) + runtime.special_files = [] + runtime.persist_files_s3 = False + runtime.capture_failed_events = False + runtime._loop = None + runtime.s3 = _FakeS3({ + ("history-bucket", "history/5491100000000.sqlite"): b"old-history", + }) + calls = [] + + class _Manager: + def import_history(self, user_id, sqlite_bytes): + calls.append(("import", user_id, sqlite_bytes)) + + def set_current_user(self, user_id): + calls.append(("current", user_id)) + + def get_messages(self, user_id): + return [] + + def add_blocks(self, content, **kwargs): + calls.append(("add_blocks", content, kwargs)) + + def export_history(self, user_id): + calls.append(("export", user_id)) + return b"history-with-outgoing" + + class _Bot: + async def process_webhook_update(self, update): + raise AssertionError("outgoing history must not process a webhook") + + runtime.manager = _Manager() + runtime.bot_instance = _Bot() + runtime.initialize_system = lambda: None + runtime.try_acquire_user_lock = lambda chat_id: True + runtime.release_user_lock = lambda chat_id: calls.append(("release", chat_id)) + + response = MawsRuntime.handle_apigw_event( + runtime, + {"maws_job": { + "provider": "history", + "chat_id": "5491100000000", + "event_id": "wamid.outbound-1", + "operation": "append_history", + "payload": {"content": "Hola desde el bulk"}, + }}, + _Context(), + ) + + assert response["statusCode"] == 200 + assert ("add_blocks", [{ + "type": "text", + "content": {"response": "Hola desde el bulk"}, + "metadata": {"maws_event_id": "wamid.outbound-1"}, + }], { + "role": "assistant", + "msg_type": "assistant", + "user_id": "5491100000000", + }) in calls + assert ("export", "5491100000000") in calls + assert ("release", "5491100000000") in calls + assert runtime.s3.puts[("history-bucket", "history/5491100000000.sqlite")] == b"history-with-outgoing" + + +def test_history_append_deduplicates_existing_event_id(): + runtime = object.__new__(MawsRuntime) + runtime.manager = types.SimpleNamespace(get_messages=lambda user_id: [{ + "message": [{"type": "text", "content": {"response": "sent"}, + "metadata": {"maws_event_id": "evt-1"}}] + }]) + + assert MawsRuntime._history_contains_external_event(runtime, "user-1", "evt-1") + assert not MawsRuntime._history_contains_external_event(runtime, "user-1", "evt-2") + + def test_optional_user_file_s3_persistence_round_trips(workspace_tmp_path): runtime = object.__new__(MawsRuntime) runtime.bucket_name = "history-bucket"