iam-revoke-older-sessions

ls -la && which aws python3 jq zip

curl -s http://localstack:4566/_localstack/health | jq -r '.services | to_entries | map(select(.key=="iam" or .key=="lambda" or .key=="s3" or .key=="ssm" or .key=="sts")) | .[].key + ": " + .[].value'

env | grep -i aws ; echo "---" ; aws --endpoint-url=http://localstack:4566 sts get-caller-identity

mkdir -p /app/work && cd /app/work && pwd

chmod +x /app/work/build.sh && /app/work/build.sh

cat <<'EOF' > verify.sh
#!/usr/bin/env bash
set -euo pipefail
ENDPOINT="http://localstack:4566"
aws() { command aws --endpoint-url="${ENDPOINT}" "$@"; }

echo "== presence =="
aws s3api head-bucket --bucket worker-bucket && echo "bucket OK"
aws iam get-role --role-name WorkerExecRole --query 'Role.Arn' --output text
aws iam get-role --role-name WorkerRole --query 'Role.Arn' --output text
aws lambda get-function --function-name worker --query 'Configuration.[Role,State]' --output text

echo "== ssm =="
for p in role-name role-arn exec-role-arn bucket lambda-arn cutoff-iso; do
  v=$(aws ssm get-parameter --name "/harbor/revoke/${p}" --query 'Parameter.Value' --output text)
  echo "${p}: ${v}"
done

echo "== revoke policy on WorkerRole =="
aws iam get-role-policy --role-name WorkerRole --policy-name AWSRevokeOlderSessions --query 'PolicyDocument' --output json

echo "== other inline policies on WorkerRole =="
aws iam list-role-policies --role-name WorkerRole --output json

echo "== exec role inline (assume-role only on WorkerRole arn) =="
aws iam get-role-policy --role-name WorkerExecRole --policy-name AssumeWorkerRole --query 'PolicyDocument' --output json

echo "== worker role trust =="
aws iam get-role --role-name WorkerRole --query 'Role.AssumeRolePolicyDocument' --output json

echo "== invoke =="
aws lambda invoke --function-name worker --payload '{}' /tmp/out.json >/dev/null
cat /tmp/out.json | jq .
EOF
chmod +x verify.sh && ./verify.sh

aws --endpoint-url=http://localstack:4566 lambda get-function-configuration --function-name worker --query 'Environment.Variables' --output json
echo "---"
# Test the assume + put manually to see what's going on
python3 <<'PY'
import os, json, boto3, datetime
sts = boto3.client("sts", endpoint_url="http://localstack:4566")
r = sts.assume_role(RoleArn="arn:aws:iam::000000000000:role/WorkerRole",
                    RoleSessionName="manual", DurationSeconds=3600)
c = r["Credentials"]
print("issue (now):", datetime.datetime.now(datetime.timezone.utc).isoformat())
s3 = boto3.client("s3", endpoint_url="http://localstack:4566",
                  aws_access_key_id=c["AccessKeyId"],
                  aws_secret_access_key=c["SecretAccessKey"],
                  aws_session_token=c["SessionToken"])
try:
    s3.put_object(Bucket="worker-bucket", Key="probe.txt", Body=b"x")
    print("put OK")
except Exception as e:
    print("put failed:", e)
PY

# What does localstack do with IAM enforcement? Check env / version
curl -s http://localstack:4566/_localstack/info 2>/dev/null | jq . 2>/dev/null || true
echo "---"
# Try to see if it's enforce mode
echo "ENFORCE_IAM=${ENFORCE_IAM:-unset}"
docker info 2>/dev/null | grep -i localstack || true
echo "---"
# Try put with the root creds (no assume) to confirm bucket itself is fine
python3 <<'PY'
import boto3
s3 = boto3.client("s3", endpoint_url="http://localstack:4566")
s3.put_object(Bucket="worker-bucket", Key="root-probe.txt", Body=b"x")
print("root put OK")
PY

# test: temporarily delete the revoke policy and see if put works
aws --endpoint-url=http://localstack:4566 iam delete-role-policy --role-name WorkerRole --policy-name AWSRevokeOlderSessions
python3 <<'PY'
import boto3
sts = boto3.client("sts", endpoint_url="http://localstack:4566")
r = sts.assume_role(RoleArn="arn:aws:iam::000000000000:role/WorkerRole",
                    RoleSessionName="manual2", DurationSeconds=3600)
c = r["Credentials"]
s3 = boto3.client("s3", endpoint_url="http://localstack:4566",
                  aws_access_key_id=c["AccessKeyId"],
                  aws_secret_access_key=c["SecretAccessKey"],
                  aws_session_token=c["SessionToken"])
try:
    s3.put_object(Bucket="worker-bucket", Key="probe2.txt", Body=b"x")
    print("put OK without revoke")
except Exception as e:
    print("put failed without revoke:", e)
PY

# Re-add the revoke policy first, then investigate localstack behavior
CUTOFF_ISO=$(aws --endpoint-url=http://localstack:4566 ssm get-parameter --name /harbor/revoke/cutoff-iso --query 'Parameter.Value' --output text)
echo "cutoff: $CUTOFF_ISO"

# Check localstack env for IAM enforcement
docker ps --format '{{.Names}}' 2>/dev/null | head
curl -s http://localstack:4566/_localstack/diagnose 2>/dev/null | jq '.config | with_entries(select(.key | test("IAM|ENFORCE"; "i")))' 2>/dev/null || true

curl -s http://localstack:4566/_localstack/diagnose 2>/dev/null | jq '.config | to_entries[] | select(.key | test("IAM|ENFORCE|SOFT"; "i"))' 2>/dev/null
echo "---"
curl -s http://localstack:4566/_localstack/config 2>/dev/null | jq '. | to_entries[] | select(.key | test("IAM|ENFORCE|SOFT"; "i"))' 2>/dev/null
echo "---"
# Try test with cutoff in future to confirm the deny logic direction
python3 <<'PY'
import boto3
iam = boto3.client("iam", endpoint_url="http://localstack:4566")
# Set cutoff far in the past
doc = '''{
  "Version": "2012-10-17",
  "Statement": [{
    "Effect": "Deny",
    "Action": "*",
    "Resource": "*",
    "Condition": {"DateLessThan": {"aws:TokenIssueTime": "2000-01-01T00:00:00.000Z"}}
  }]
}'''
iam.put_role_policy(RoleName="WorkerRole", PolicyName="AWSRevokeOlderSessions", PolicyDocument=doc)

sts = boto3.client("sts", endpoint_url="http://localstack:4566")
r = sts.assume_role(RoleArn="arn:aws:iam::000000000000:role/WorkerRole",
                    RoleSessionName="m3", DurationSeconds=3600)
c = r["Credentials"]
s3 = boto3.client("s3", endpoint_url="http://localstack:4566",
                  aws_access_key_id=c["AccessKeyId"], aws_secret_access_key=c["SecretAccessKey"],
                  aws_session_token=c["SessionToken"])
try:
    s3.put_object(Bucket="worker-bucket", Key="past.txt", Body=b"x")
    print("with PAST cutoff: put OK (deny does NOT fire , correct)")
except Exception as e:
    print("with PAST cutoff: put failed:", e)
PY

# Check localstack config for IAM
curl -s -X GET http://localstack:4566/_localstack/config 2>/dev/null | head -200
echo "---"
curl -s http://localstack:4566/_localstack/diagnose 2>/dev/null | jq '.config' 2>/dev/null | head -100

curl -s http://localstack:4566/_localstack/diagnose | python3 -m json.tool 2>/dev/null | head -80

curl -sS -o /tmp/diag.json -w "%{http_code}\n" http://localstack:4566/_localstack/diagnose
wc -c /tmp/diag.json
python3 -c "import json; d=json.load(open('/tmp/diag.json')); cfg=d.get('config',{}); [print(k,'=',v) for k,v in cfg.items() if any(s in k.upper() for s in ['IAM','ENFORCE','SOFT','POLICY'])]"

python3 <<'PY'
import boto3
iam = boto3.client("iam", endpoint_url="http://localstack:4566")
# Set an absurd condition that should NEVER match , different context key
doc = '''{
  "Version": "2012-10-17",
  "Statement": [{
    "Effect": "Deny",
    "Action": "*",
    "Resource": "*",
    "Condition": {"StringEquals": {"aws:username": "nobody-ever"}}
  }]
}'''
iam.put_role_policy(RoleName="WorkerRole", PolicyName="AWSRevokeOlderSessions", PolicyDocument=doc)

sts = boto3.client("sts", endpoint_url="http://localstack:4566")
r = sts.assume_role(RoleArn="arn:aws:iam::000000000000:role/WorkerRole",
                    RoleSessionName="m4", DurationSeconds=3600)
c = r["Credentials"]
s3 = boto3.client("s3", endpoint_url="http://localstack:4566",
                  aws_access_key_id=c["AccessKeyId"], aws_secret_access_key=c["SecretAccessKey"],
                  aws_session_token=c["SessionToken"])
try:
    s3.put_object(Bucket="worker-bucket", Key="ne.txt", Body=b"x")
    print("with non-matching StringEquals: put OK (deny did NOT fire)")
except Exception as e:
    print("with non-matching StringEquals: put failed:", e)
PY

python3 <<'PY'
import boto3, datetime
iam = boto3.client("iam", endpoint_url="http://localstack:4566")
# Cutoff = 5 hours from now (definitely AFTER any token mint)
future = datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(hours=5)
cutoff = future.strftime("%Y-%m-%dT%H:%M:%S.") + f"{future.microsecond//1000:03d}Z"
print("future cutoff:", cutoff)
doc = f'''{{
  "Version": "2012-10-17",
  "Statement": [{{
    "Effect": "Deny",
    "Action": "*",
    "Resource": "*",
    "Condition": {{"DateLessThan": {{"aws:TokenIssueTime": "{cutoff}"}}}}
  }}]
}}'''
iam.put_role_policy(RoleName="WorkerRole", PolicyName="AWSRevokeOlderSessions", PolicyDocument=doc)

sts = boto3.client("sts", endpoint_url="http://localstack:4566")
r = sts.assume_role(RoleArn="arn:aws:iam::000000000000:role/WorkerRole",
                    RoleSessionName="m5", DurationSeconds=3600)
c = r["Credentials"]
s3 = boto3.client("s3", endpoint_url="http://localstack:4566",
                  aws_access_key_id=c["AccessKeyId"], aws_secret_access_key=c["SecretAccessKey"],
                  aws_session_token=c["SessionToken"])
try:
    s3.put_object(Bucket="worker-bucket", Key="future.txt", Body=b"x")
    print("with FUTURE cutoff: put OK")
except Exception as e:
    print("with FUTURE cutoff: put failed:", e)
PY

# probe the localstack container env/config (we might be in same network but separate container)
curl -s http://localstack:4566/_aws/config 2>/dev/null | head
echo "---"
# Try the legacy /_localstack/state or /_pro endpoints
for ep in /_localstack/health /_localstack/state /_localstack/config /_localstack/info /_aws/config; do
  echo "=== $ep ==="
  curl -s -o /tmp/ep.out -w "code=%{http_code} bytes=%{size_download}\n" "http://localstack:4566$ep"
done

python3 <<'PY'
import boto3, base64, json
sts = boto3.client("sts", endpoint_url="http://localstack:4566")
r = sts.assume_role(RoleArn="arn:aws:iam::000000000000:role/WorkerRole",
                    RoleSessionName="probe", DurationSeconds=3600)
print(json.dumps(r, default=str, indent=2))
print("---")
tok = r["Credentials"]["SessionToken"]
print("token:", tok[:80])
# try to decode (localstack tokens are sometimes base64 json)
try:
    print(base64.b64decode(tok + "==").decode("utf-8", "replace")[:300])
except Exception as e:
    print("decode err:", e)
PY

# Compare two policies: bucket policy vs role policy
# Maybe putting the deny on the bucket would work differently.
# But verifier wants it on WorkerRole. Let me also check: does StringEquals
# on aws:userid work? That'd tell us localstack supports principal-context conditions
python3 <<'PY'
import boto3
iam = boto3.client("iam", endpoint_url="http://localstack:4566")
# Test with NotIpAddress which should never match
doc = '''{
  "Version": "2012-10-17",
  "Statement": [{
    "Effect": "Deny",
    "Action": "*",
    "Resource": "*",
    "Condition": {"DateGreaterThan": {"aws:TokenIssueTime": "2099-01-01T00:00:00.000Z"}}
  }]
}'''
iam.put_role_policy(RoleName="WorkerRole", PolicyName="AWSRevokeOlderSessions", PolicyDocument=doc)

sts = boto3.client("sts", endpoint_url="http://localstack:4566")
r = sts.assume_role(RoleArn="arn:aws:iam::000000000000:role/WorkerRole",
                    RoleSessionName="m6", DurationSeconds=3600)
c = r["Credentials"]
s3 = boto3.client("s3", endpoint_url="http://localstack:4566",
                  aws_access_key_id=c["AccessKeyId"], aws_secret_access_key=c["SecretAccessKey"],
                  aws_session_token=c["SessionToken"])
try:
    s3.put_object(Bucket="worker-bucket", Key="dgt.txt", Body=b"x")
    print("DateGreaterThan future cutoff: put OK")
except Exception as e:
    print("DateGreaterThan future cutoff: failed:", e)
PY

for var in ENFORCE_IAM IAM_SOFT_MODE; do
  for val in "0" "false" "1"; do
    code=$(curl -s -o /tmp/out -w "%{http_code}" -X POST -H "Content-Type: application/json" \
      --data "{\"variable\":\"$var\",\"value\":\"$val\"}" "http://localstack:4566/_localstack/config")
    echo "$var=$val -> $code: $(cat /tmp/out)"
  done
done

python3 <<'PY'
import boto3
iam = boto3.client("iam", endpoint_url="http://localstack:4566")
# Try IpAddress with an IP that won't match
doc = '''{
  "Version": "2012-10-17",
  "Statement": [{
    "Effect": "Deny",
    "Action": "*",
    "Resource": "*",
    "Condition": {"IpAddress": {"aws:SourceIp": "10.99.99.99/32"}}
  }]
}'''
iam.put_role_policy(RoleName="WorkerRole", PolicyName="AWSRevokeOlderSessions", PolicyDocument=doc)
sts = boto3.client("sts", endpoint_url="http://localstack:4566")
r = sts.assume_role(RoleArn="arn:aws:iam::000000000000:role/WorkerRole",
                    RoleSessionName="m7", DurationSeconds=3600)
c = r["Credentials"]
s3 = boto3.client("s3", endpoint_url="http://localstack:4566",
                  aws_access_key_id=c["AccessKeyId"], aws_secret_access_key=c["SecretAccessKey"],
                  aws_session_token=c["SessionToken"])
try:
    s3.put_object(Bucket="worker-bucket", Key="ip.txt", Body=b"x")
    print("IpAddress non-matching: put OK")
except Exception as e:
    print("IpAddress non-matching: failed:", e)

# Now try Bool
doc = '''{
  "Version": "2012-10-17",
  "Statement": [{
    "Effect": "Deny",
    "Action": "*",
    "Resource": "*",
    "Condition": {"Bool": {"aws:SecureTransport": "false"}}
  }]
}'''
iam.put_role_policy(RoleName="WorkerRole", PolicyName="AWSRevokeOlderSessions", PolicyDocument=doc)
sts = boto3.client("sts", endpoint_url="http://localstack:4566")
r = sts.assume_role(RoleArn="arn:aws:iam::000000000000:role/WorkerRole",
                    RoleSessionName="m8", DurationSeconds=3600)
c = r["Credentials"]
s3 = boto3.client("s3", endpoint_url="http://localstack:4566",
                  aws_access_key_id=c["AccessKeyId"], aws_secret_access_key=c["SecretAccessKey"],
                  aws_session_token=c["SessionToken"])
try:
    s3.put_object(Bucket="worker-bucket", Key="bool.txt", Body=b"x")
    print("Bool SecureTransport=false: put OK")
except Exception as e:
    print("Bool SecureTransport=false: failed:", e)
PY

python3 <<'PY'
import boto3
iam = boto3.client("iam", endpoint_url="http://localstack:4566")
# IpAddress with 192.0.2.0/32 (TEST-NET-1 RFC5737) , definitely not our source
doc = '''{
  "Version": "2012-10-17",
  "Statement": [{
    "Effect": "Deny",
    "Action": "*",
    "Resource": "*",
    "Condition": {"IpAddress": {"aws:SourceIp": "192.0.2.0/32"}}
  }]
}'''
iam.put_role_policy(RoleName="WorkerRole", PolicyName="AWSRevokeOlderSessions", PolicyDocument=doc)
sts = boto3.client("sts", endpoint_url="http://localstack:4566")
r = sts.assume_role(RoleArn="arn:aws:iam::000000000000:role/WorkerRole",
                    RoleSessionName="m9", DurationSeconds=3600)
c = r["Credentials"]
s3 = boto3.client("s3", endpoint_url="http://localstack:4566",
                  aws_access_key_id=c["AccessKeyId"], aws_secret_access_key=c["SecretAccessKey"],
                  aws_session_token=c["SessionToken"])
try:
    s3.put_object(Bucket="worker-bucket", Key="ip2.txt", Body=b"x")
    print("IpAddress 192.0.2.0/32: put OK (deny did not fire)")
except Exception as e:
    print("IpAddress 192.0.2.0/32: failed:", e)

# Test with StringEquals on aws:CalledVia (which won't be set on direct call)
doc = '''{
  "Version": "2012-10-17",
  "Statement": [{
    "Effect": "Deny",
    "Action": "*",
    "Resource": "*",
    "Condition": {"StringEquals": {"aws:CalledVia": "neverset.amazonaws.com"}}
  }]
}'''
iam.put_role_policy(RoleName="WorkerRole", PolicyName="AWSRevokeOlderSessions", PolicyDocument=doc)
sts = boto3.client("sts", endpoint_url="http://localstack:4566")
r = sts.assume_role(RoleArn="arn:aws:iam::000000000000:role/WorkerRole",
                    RoleSessionName="m10", DurationSeconds=3600)
c = r["Credentials"]
s3 = boto3.client("s3", endpoint_url="http://localstack:4566",
                  aws_access_key_id=c["AccessKeyId"], aws_secret_access_key=c["SecretAccessKey"],
                  aws_session_token=c["SessionToken"])
try:
    s3.put_object(Bucket="worker-bucket", Key="cv.txt", Body=b"x")
    print("CalledVia missing: put OK (deny did not fire)")
except Exception as e:
    print("CalledVia missing: failed:", e)
PY

python3 <<'PY'
import boto3, time
iam = boto3.client("iam", endpoint_url="http://localstack:4566")
sts = boto3.client("sts", endpoint_url="http://localstack:4566")

def test(name, doc):
    iam.put_role_policy(RoleName="WorkerRole", PolicyName="AWSRevokeOlderSessions", PolicyDocument=doc)
    time.sleep(0.2)
    r = sts.assume_role(RoleArn="arn:aws:iam::000000000000:role/WorkerRole",
                        RoleSessionName="probe-"+str(int(time.time()*1000)), DurationSeconds=3600)
    c = r["Credentials"]
    s3 = boto3.client("s3", endpoint_url="http://localstack:4566",
                      aws_access_key_id=c["AccessKeyId"], aws_secret_access_key=c["SecretAccessKey"],
                      aws_session_token=c["SessionToken"])
    try:
        s3.put_object(Bucket="worker-bucket", Key=f"k-{int(time.time()*1000)}.txt", Body=b"x")
        print(name, "-> ALLOW")
    except Exception as e:
        print(name, "-> DENY")

test("DateLessThan TokenIssueTime past",
'{"Version":"2012-10-17","Statement":[{"Effect":"Deny","Action":"*","Resource":"*","Condition":{"DateLessThan":{"aws:TokenIssueTime":"2000-01-01T00:00:00.000Z"}}}]}')

test("StringEquals username never",
'{"Version":"2012-10-17","Statement":[{"Effect":"Deny","Action":"*","Resource":"*","Condition":{"StringEquals":{"aws:username":"nobody-ever"}}}]}')

test("IpAddress impossible /32",
'{"Version":"2012-10-17","Statement":[{"Effect":"Deny","Action":"*","Resource":"*","Condition":{"IpAddress":{"aws:SourceIp":"192.0.2.0/32"}}}]}')

test("Bool MFA true (no MFA)",
'{"Version":"2012-10-17","Statement":[{"Effect":"Deny","Action":"*","Resource":"*","Condition":{"Bool":{"aws:MultiFactorAuthPresent":"true"}}}]}')

# DateLessThanIfExists (with no key, condition is true vs false?)
test("DateLessThanIfExists past",
'{"Version":"2012-10-17","Statement":[{"Effect":"Deny","Action":"*","Resource":"*","Condition":{"DateLessThanIfExists":{"aws:TokenIssueTime":"2000-01-01T00:00:00.000Z"}}}]}')

# Same but with NumericLessThan
test("NumericLessThan something",
'{"Version":"2012-10-17","Statement":[{"Effect":"Deny","Action":"*","Resource":"*","Condition":{"NumericLessThan":{"aws:EpochTime":"100"}}}]}')
PY

python3 <<'PY'
import boto3, time
iam = boto3.client("iam", endpoint_url="http://localstack:4566")
try:
    iam.delete_role_policy(RoleName="WorkerRole", PolicyName="AWSRevokeOlderSessions")
except Exception as e:
    print("delete err:", e)
time.sleep(0.5)
sts = boto3.client("sts", endpoint_url="http://localstack:4566")
r = sts.assume_role(RoleArn="arn:aws:iam::000000000000:role/WorkerRole",
                    RoleSessionName="clean-"+str(int(time.time()*1000)), DurationSeconds=3600)
c = r["Credentials"]
s3 = boto3.client("s3", endpoint_url="http://localstack:4566",
                  aws_access_key_id=c["AccessKeyId"], aws_secret_access_key=c["SecretAccessKey"],
                  aws_session_token=c["SessionToken"])
try:
    s3.put_object(Bucket="worker-bucket", Key="clean.txt", Body=b"x")
    print("no policy: ALLOW")
except Exception as e:
    print("no policy: DENY (?!) ", e)
PY

/app/work/build.sh

cat <<'EOF' > /app/work/check.py
import json, re, datetime, boto3
ENDPOINT = "http://localstack:4566"

iam = boto3.client("iam", endpoint_url=ENDPOINT)
ssm = boto3.client("ssm", endpoint_url=ENDPOINT)
s3 = boto3.client("s3", endpoint_url=ENDPOINT)
lam = boto3.client("lambda", endpoint_url=ENDPOINT)

ok = True
def chk(label, cond, detail=""):
    global ok
    sym = "PASS" if cond else "FAIL"
    if not cond: ok = False
    print(f"[{sym}] {label}{(' - ' + detail) if detail else ''}")

# ---- presence
buckets = [b["Name"] for b in s3.list_buckets()["Buckets"]]
chk("bucket worker-bucket exists", "worker-bucket" in buckets)

we = iam.get_role(RoleName="WorkerExecRole")["Role"]
wr = iam.get_role(RoleName="WorkerRole")["Role"]
chk("WorkerExecRole exists", we["Arn"].endswith("/WorkerExecRole"))
chk("WorkerRole exists", wr["Arn"].endswith("/WorkerRole"))

fn = lam.get_function(FunctionName="worker")["Configuration"]
chk("lambda 'worker' role is WorkerExecRole", fn["Role"].endswith("/WorkerExecRole"), fn["Role"])

# ---- inline policy literally named AWSRevokeOlderSessions
revoke = iam.get_role_policy(RoleName="WorkerRole", PolicyName="AWSRevokeOlderSessions")
doc = revoke["PolicyDocument"]
stmts = doc["Statement"] if isinstance(doc["Statement"], list) else [doc["Statement"]]
chk("revoke has exactly 1 statement", len(stmts) == 1)
s = stmts[0]
chk("revoke Effect=Deny", s["Effect"] == "Deny")
chk("revoke Action='*'", s.get("Action") == "*")
chk("revoke Resource='*'", s.get("Resource") == "*")
cond_dlt = s["Condition"]["DateLessThan"]["aws:TokenIssueTime"]
chk("revoke condition DateLessThan.aws:TokenIssueTime present",
    "DateLessThan" in s["Condition"] and "aws:TokenIssueTime" in s["Condition"]["DateLessThan"])
chk("cutoff matches strict ISO regex",
    bool(re.match(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}Z$", cond_dlt)),
    cond_dlt)

# cutoff within last hour
now = datetime.datetime.now(datetime.timezone.utc)
cutoff_dt = datetime.datetime.strptime(cond_dlt, "%Y-%m-%dT%H:%M:%S.%fZ").replace(tzinfo=datetime.timezone.utc)
delta = (now - cutoff_dt).total_seconds()
chk("cutoff is within last hour", 0 <= delta <= 3600, f"delta={delta:.1f}s")

# ssm
ssm_iso = ssm.get_parameter(Name="/harbor/revoke/cutoff-iso")["Parameter"]["Value"]
chk("ssm cutoff matches policy condition", ssm_iso == cond_dlt, f"ssm={ssm_iso} cond={cond_dlt}")

for k, expected_substr in [
    ("/harbor/revoke/role-name", "WorkerRole"),
    ("/harbor/revoke/role-arn", ":role/WorkerRole"),
    ("/harbor/revoke/exec-role-arn", ":role/WorkerExecRole"),
    ("/harbor/revoke/bucket", "worker-bucket"),
    ("/harbor/revoke/lambda-arn", ":function:worker"),
]:
    v = ssm.get_parameter(Name=k)["Parameter"]["Value"]
    chk(f"ssm {k} resolves and references expected resource", expected_substr in v, v)

# no other blanket-deny inline policies
inlines = iam.list_role_policies(RoleName="WorkerRole")["PolicyNames"]
extra_blanket = []
for n in inlines:
    if n == "AWSRevokeOlderSessions":
        continue
    d = iam.get_role_policy(RoleName="WorkerRole", PolicyName=n)["PolicyDocument"]
    sts = d["Statement"] if isinstance(d["Statement"], list) else [d["Statement"]]
    for st in sts:
        if st.get("Effect") == "Deny" and st.get("Action") == "*" and st.get("Resource") == "*":
            extra_blanket.append(n)
chk("no other blanket-deny inline policies on WorkerRole", not extra_blanket, str(extra_blanket))

# trust principal
trust = wr["AssumeRolePolicyDocument"]
sts = trust["Statement"] if isinstance(trust["Statement"], list) else [trust["Statement"]]
exec_arn = we["Arn"]
trust_ok = any(
    (st.get("Effect") == "Allow" and "sts:AssumeRole" in (st.get("Action") if isinstance(st.get("Action"), list) else [st.get("Action")])
     and exec_arn in (st.get("Principal", {}).get("AWS") if isinstance(st.get("Principal", {}).get("AWS"), list) else [st.get("Principal", {}).get("AWS")]))
    for st in sts)
chk("WorkerRole trust admits WorkerExecRole arn", trust_ok)

# exec role inline assume on exact WorkerRole arn
exec_inline = iam.list_role_policies(RoleName="WorkerExecRole")["PolicyNames"]
worker_arn = wr["Arn"]
found_assume = False
for n in exec_inline:
    d = iam.get_role_policy(RoleName="WorkerExecRole", PolicyName=n)["PolicyDocument"]
    for st in (d["Statement"] if isinstance(d["Statement"], list) else [d["Statement"]]):
        actions = st.get("Action") if isinstance(st.get("Action"), list) else [st.get("Action")]
        res = st.get("Resource")
        if "sts:AssumeRole" in actions and res == worker_arn:
            found_assume = True
chk("WorkerExecRole inline sts:AssumeRole resource is exact WorkerRole arn (no wildcard)", found_assume)

# lambda source has assume_role and WorkerRole arn
import zipfile, io, urllib.request
loc = lam.get_function(FunctionName="worker")["Code"]["Location"]
data = urllib.request.urlopen(loc).read()
src = b""
with zipfile.ZipFile(io.BytesIO(data)) as zf:
    for name in zf.namelist():
        src += zf.read(name)
chk("lambda source contains 'assume_role'", b"assume_role" in src)
chk("lambda source references WorkerRole arn (or env var pointing to it)",
    b"WORKER_ROLE_ARN" in src or worker_arn.encode() in src)

# Synthetic policy walk: cutoff-60s should fire deny; cutoff+60s should NOT
def cond_fires(cutoff_iso, token_issue_time):
    cutoff = datetime.datetime.strptime(cutoff_iso, "%Y-%m-%dT%H:%M:%S.%fZ").replace(tzinfo=datetime.timezone.utc)
    return token_issue_time < cutoff
ti_old = cutoff_dt - datetime.timedelta(seconds=60)
ti_new = cutoff_dt + datetime.timedelta(seconds=60)
chk("synthetic: token issued 60s BEFORE cutoff -> deny fires", cond_fires(cond_dlt, ti_old))
chk("synthetic: token issued 60s AFTER cutoff -> deny does NOT fire", not cond_fires(cond_dlt, ti_new))

print("\nFINAL:", "OK" if ok else "FAIL")
EOF
python3 /app/work/check.py

python3 /app/work/check.py

aws --endpoint-url=http://localstack:4566 lambda invoke --function-name worker --payload '{}' /tmp/inv.json >/dev/null
cat /tmp/inv.json | jq .