Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions .cspell/custom-words.txt
Original file line number Diff line number Diff line change
Expand Up @@ -118,6 +118,7 @@ okhttp
opentelemetry
otelgrpc
otelhttp
otherpisp
Otherville
OURCYGPATTERN
Palo
Expand All @@ -127,6 +128,10 @@ Payoneer
paypal
Payplug
pids
PISP
Pisp
Pisps
pisp
pmezard
proguard
Proguard
Expand All @@ -149,6 +154,7 @@ ropeproject
RPCURL
Rulebook
screenreaders
SECP
setlocal
sharedpref
Shopcider
Expand Down
27 changes: 22 additions & 5 deletions code/sdk/python/ap2/sdk/constraints.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,7 @@
from __future__ import annotations

from abc import ABC, abstractmethod
from datetime import datetime, timedelta
from datetime import UTC, datetime, timedelta

from ap2.sdk.generated.open_checkout_mandate import (
AllowedMerchants,
Expand Down Expand Up @@ -316,8 +316,13 @@ def evaluate(
class ExecutionDateEvaluator(PaymentConstraintEvaluator):
"""Evaluates if the execution date is within the allowed window."""

def __init__(self, constraint: ExecutionDate):
def __init__(
self,
constraint: ExecutionDate,
current_time: datetime | None = None,
):
self.constraint = constraint
self.current_time = current_time

def evaluate(
self,
Expand All @@ -326,7 +331,12 @@ def evaluate(
) -> list[str]:
exec_date = closed_mandate.execution_date
if not exec_date:
return []
effective_time = self.current_time or datetime.now(UTC)
if effective_time.tzinfo is None:
effective_time = effective_time.replace(tzinfo=UTC)
exec_date = effective_time.astimezone(UTC).isoformat().replace(
'+00:00', 'Z'
)

violations = []
if (
Expand Down Expand Up @@ -357,6 +367,7 @@ def create_payment_evaluator( # noqa: PLR0911
| PaymentReference
),
mandate_context: MandateContext | None = None,
current_time: datetime | None = None,
) -> PaymentConstraintEvaluator:
"""Factory: create the appropriate evaluator for a payment constraint."""
if isinstance(constraint, AmountRange):
Expand All @@ -374,7 +385,7 @@ def create_payment_evaluator( # noqa: PLR0911
if isinstance(constraint, Budget):
return BudgetEvaluator(constraint, mandate_context)
if isinstance(constraint, ExecutionDate):
return ExecutionDateEvaluator(constraint)
return ExecutionDateEvaluator(constraint, current_time)
raise ValueError(f'Unknown payment constraint type: {type(constraint)}')


Expand Down Expand Up @@ -491,6 +502,7 @@ def check_payment_constraints(
closed_payment: PaymentMandate,
open_checkout_hash: str | None = None,
mandate_context: MandateContext | None = None,
current_time: datetime | None = None,
) -> list[str]:
"""Verify the closed payment satisfies open mandate constraints.

Expand All @@ -503,6 +515,7 @@ def check_payment_constraints(
open_checkout_hash: The hash of the open checkout mandate, required for
`PaymentReference` constraints.
mandate_context: Aggregated usage context for the mandate.
current_time: Trusted current time used for immediate payments.

Returns:
A list of strings, where each string describes a violation of the
Expand Down Expand Up @@ -533,7 +546,11 @@ def check_payment_constraints(
)

for constraint in open_mandate.constraints:
evaluator = create_payment_evaluator(constraint, mandate_context)
evaluator = create_payment_evaluator(
constraint,
mandate_context,
current_time,
)
violations.extend(
evaluator.evaluate(closed_payment, open_checkout_hash)
)
Expand Down
4 changes: 4 additions & 0 deletions code/sdk/python/ap2/sdk/payment_mandate_chain.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
from __future__ import annotations

from dataclasses import dataclass
from datetime import datetime
from typing import Any

from ap2.sdk.constraints import MandateContext, check_payment_constraints
Expand Down Expand Up @@ -40,6 +41,7 @@ def verify(
expected_transaction_id: str | None = None,
expected_open_checkout_hash: str | None = None,
mandate_context: MandateContext | None = None,
current_time: datetime | None = None,
) -> list[str]:
"""Verifies the constraints of the payment mandate chain.

Expand All @@ -49,6 +51,7 @@ def verify(
expected_open_checkout_hash: Optional checkout hash to check against
the open mandate's checkout_reference.
mandate_context: Aggregated usage context for the mandate.
current_time: Trusted current time used for immediate payments.

Returns:
A list of strings describing any violations found.
Expand All @@ -71,6 +74,7 @@ def verify(
self.closed_mandate,
open_checkout_hash=expected_open_checkout_hash,
mandate_context=mandate_context,
current_time=current_time,
)
if (
expected_transaction_id is not None
Expand Down
51 changes: 49 additions & 2 deletions code/sdk/python/ap2/tests/constraints_tests.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
"""Tests for centralized constraint checking (ap2.sdk.constraints)."""

import time
from datetime import UTC, datetime

import pytest

Expand Down Expand Up @@ -534,15 +535,61 @@ def test_payment_execution_date_after_window():
assert any('after allowed window' in v for v in violations)


def test_payment_execution_date_missing_passes():
"""Missing execution date (immediate) passes."""
def test_payment_execution_date_missing_uses_current_time_within_window():
"""Immediate execution within the authorized window passes."""
violations = check_payment_constraints(
_open_payment(
constraints=[
ExecutionDate(not_before='2025-01-01', not_after='2025-12-31'),
]
),
_closed_payment(execution_date=None),
current_time=datetime(2025, 6, 1, tzinfo=UTC),
)
assert violations == []


def test_payment_execution_date_missing_before_window():
"""Immediate execution before the authorized window is a violation."""
violations = check_payment_constraints(
_open_payment(
constraints=[
ExecutionDate(not_before='2025-01-01', not_after='2025-12-31'),
]
),
_closed_payment(execution_date=None),
current_time=datetime(2024, 12, 31, tzinfo=UTC),
)
assert any('before allowed window' in v for v in violations)


def test_payment_execution_date_missing_after_window():
"""Immediate execution after the authorized window is a violation."""
violations = check_payment_constraints(
_open_payment(
constraints=[
ExecutionDate(not_before='2025-01-01', not_after='2025-12-31'),
]
),
_closed_payment(execution_date=None),
current_time=datetime(2026, 1, 1, tzinfo=UTC),
)
assert any('after allowed window' in v for v in violations)


def test_payment_execution_date_missing_at_exact_boundary():
"""Immediate execution at an ISO 8601 boundary passes."""
violations = check_payment_constraints(
_open_payment(
constraints=[
ExecutionDate(
not_before='2025-06-01T00:00:00Z',
not_after='2025-06-01T00:00:00Z',
),
]
),
_closed_payment(execution_date=None),
current_time=datetime(2025, 6, 1, tzinfo=UTC),
)
assert violations == []

Expand Down
Loading