|
"""Ad-hoc checks for the ported OTP mailbox + login config (no network, no browser)."""
|
|
import sys
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
sys.path.insert(0, ".")
|
|
sys.path.insert(0, "app/phase_a_fetch")
|
|
|
|
from shared.otp_mailbox import ( # noqa: E402
|
|
FrappeOtpMailbox,
|
|
OtpMailboxError,
|
|
OtpRequest,
|
|
MailboxBaseline,
|
|
extract_forwarded_sent_time,
|
|
extract_otp,
|
|
extract_user_id,
|
|
)
|
|
from shared.nsdl_login import NsdlLoginConfigError, build_nsdl_login_cfg # noqa: E402
|
|
|
|
UTC = timezone.utc
|
|
IST = timezone(timedelta(hours=5, minutes=30))
|
|
failures = []
|
|
|
|
|
|
def check(name, cond, detail=""):
|
|
print(("PASS " if cond else "FAIL ") + name + (f" [{detail}]" if detail and not cond else ""))
|
|
if not cond:
|
|
failures.append(name)
|
|
|
|
|
|
# ---------------------------------------------------------------- pure extraction
|
|
SENT = "658204 is your One Time Password (OTP) of User ID user01 for verification on SEZ Online System. It is valid for 10 mins."
|
|
check("extract_otp NSDL sentence", extract_otp(SENT) == "658204")
|
|
check("extract_otp 'OTP is N'", extract_otp("Your OTP is 771204.") == "771204")
|
|
check("extract_otp none", extract_otp("nothing here") is None)
|
|
check("extract_otp ambiguous (two different OTPs) -> None",
|
|
extract_otp("111111 is your One Time Password ... 222222 is your One Time Password") is None)
|
|
check("extract_user_id", extract_user_id(SENT) == "user01")
|
|
fwd = "From: x Sent: Saturday, October 10, 2026 11:39:20 AM To: y Subject: SEZOnline Authentication " + SENT
|
|
when, precision = extract_forwarded_sent_time(fwd, IST)
|
|
check("forwarded Sent: parsed as UTC", when == datetime(2026, 10, 10, 6, 9, 20, tzinfo=UTC), str(when))
|
|
check("forwarded Sent: precision seconds", precision == 1)
|
|
|
|
# ---------------------------------------------------------------- fake Frappe
|
|
class Resp:
|
|
ok = True
|
|
status_code = 200
|
|
text = "ok"
|
|
headers = {}
|
|
|
|
def __init__(self, data):
|
|
self._data = data
|
|
|
|
def json(self):
|
|
return self._data
|
|
|
|
|
|
class FakeSession:
|
|
def __init__(self):
|
|
self.headers = {}
|
|
self.docs = []
|
|
|
|
def request(self, method, url, **kw):
|
|
return Resp({"data": list(self.docs)})
|
|
|
|
def post(self, url, **kw):
|
|
return Resp({"message": None})
|
|
|
|
def get(self, url, **kw):
|
|
return Resp({"message": {"time_zone": "Asia/Kolkata"}})
|
|
|
|
|
|
def mailbox(session, now):
|
|
return FrappeOtpMailbox(
|
|
base_url="https://x.example", api_key="k", api_secret="s", email_account="OTP Email",
|
|
expected_user_id="user01", frappe_time_zone="Asia/Kolkata", session=session,
|
|
sleep=lambda s: None, clock=lambda: 0.0, utc_now=lambda: now[0], trigger_pull=False,
|
|
otp_validity_seconds=600,
|
|
)
|
|
|
|
|
|
def doc(name, otp, sent_utc, user="user01", forwarded_sent_utc=None):
|
|
body = f"{otp} is your One Time Password (OTP) of User ID {user} for verification. valid for 10 mins."
|
|
if forwarded_sent_utc is not None:
|
|
ist = forwarded_sent_utc.astimezone(IST)
|
|
body = f"---------- Forwarded ---------- Sent: {ist.strftime('%A, %B %d, %Y %I:%M:%S %p')} To: a Subject: SEZOnline Authentication {body}"
|
|
return {
|
|
"name": name, "subject": "Fw: SEZOnline Authentication", "creation": sent_utc.astimezone(IST).strftime("%Y-%m-%d %H:%M:%S.%f"),
|
|
"communication_date": sent_utc.astimezone(IST).strftime("%Y-%m-%d %H:%M:%S"), "message_id": f"<{name}@x>",
|
|
"sender": "m@x", "content": body, "text_content": body,
|
|
}
|
|
|
|
|
|
T0 = datetime(2026, 10, 12, 6, 29, 39, tzinfo=UTC) # Login click (this morning's second failed run)
|
|
now = [T0]
|
|
sess = FakeSession()
|
|
box = mailbox(sess, now)
|
|
|
|
req1 = OtpRequest(sequence=1, baseline=MailboxBaseline(names=frozenset(), latest_creation=""))
|
|
req1.stamp(T0)
|
|
|
|
# A mail issued BEFORE the click (an old OTP still unread in the inbox) must be ignored.
|
|
sess.docs = [doc("OLD", "111111", T0 - timedelta(minutes=3))]
|
|
now[0] = T0 + timedelta(seconds=30)
|
|
check("mail older than the request is ignored", box.wait_for_otp(req1, 0) is None, str(box.last_scan_summary))
|
|
|
|
# The correct mail (sent just after the click) is accepted.
|
|
sess.docs.append(doc("GOOD1", "222222", T0 + timedelta(seconds=40)))
|
|
msg = box.wait_for_otp(req1, 0)
|
|
check("fresh mail for request #1 accepted", msg is not None and msg.otp == "222222")
|
|
|
|
# Resend #2 happens 95s later. The first request's mail arrives late (sent at +60s) -> must be stale for #2.
|
|
T2 = T0 + timedelta(seconds=95)
|
|
now[0] = T2
|
|
req2 = OtpRequest(sequence=2, baseline=MailboxBaseline(names=frozenset({"OLD"}), latest_creation=""),
|
|
previous_requested_at=T0)
|
|
req2.stamp(T2)
|
|
sess.docs = [doc("OLD", "111111", T0 - timedelta(minutes=3)),
|
|
doc("LATE1", "333333", T0 + timedelta(seconds=60))]
|
|
box2 = mailbox(sess, now)
|
|
check("late mail of the superseded request is rejected (this morning's failure)",
|
|
box2.wait_for_otp(req2, 0) is None and "LATE1" in box2.last_scan_summary, str(box2.last_scan_summary))
|
|
sess.docs.append(doc("GOOD2", "444444", T2 + timedelta(seconds=20)))
|
|
msg2 = box2.wait_for_otp(req2, 0)
|
|
check("mail sent after the resend is accepted", msg2 is not None and msg2.otp == "444444")
|
|
|
|
# A used OTP is never returned again.
|
|
box2.mark_used(msg2)
|
|
check("used OTP never returned again", box2.wait_for_otp(req2, 0) is None)
|
|
|
|
# Forwarded copy of an OLD NSDL mail (forward is new, original Sent is old) is rejected.
|
|
now[0] = T2 + timedelta(seconds=40)
|
|
sess.docs = [doc("FWDOLD", "555555", T2 + timedelta(seconds=30), forwarded_sent_utc=T0 - timedelta(minutes=5))]
|
|
box3 = mailbox(sess, now)
|
|
req3 = OtpRequest(sequence=2, baseline=MailboxBaseline(), previous_requested_at=None)
|
|
req3.stamp(T2)
|
|
check("forward of an older NSDL mail is rejected", box3.wait_for_otp(req3, 0) is None
|
|
and "older NSDL mail" in "".join(box3.last_scan_summary.values()), str(box3.last_scan_summary))
|
|
|
|
# Different NSDL user is ignored.
|
|
sess.docs = [doc("OTHER", "666666", T2 + timedelta(seconds=30), user="someoneelse")]
|
|
check("OTP for another NSDL User ID is ignored", box3.wait_for_otp(req3, 0) is None)
|
|
|
|
# Expired OTP rejected.
|
|
now[0] = T2 + timedelta(seconds=700)
|
|
sess.docs = [doc("EXP", "777777", T2 + timedelta(seconds=10))]
|
|
check("expired OTP rejected", mailbox(sess, now).wait_for_otp(req3, 0) is None)
|
|
|
|
# ---------------------------------------------------------------- config builder
|
|
acct = {"NSDL_USERNAME": "u", "NSDL_PASSWORD": "p", "OTP_FRAPPE_URL": "https://x", "OTP_FRAPPE_API_KEY": "k",
|
|
"OTP_FRAPPE_API_SECRET": "s", "OTP_EMAIL_ACCOUNT": "OTP Email"}
|
|
cfg = build_nsdl_login_cfg(acct, environ={})
|
|
check("defaults: wait 540, no resend on timeout, 2 submits, captcha len 5",
|
|
cfg["nsdl_otp"]["wait_seconds"] == 540 and cfg["nsdl_otp"]["resend_on_timeout"] is False
|
|
and cfg["nsdl_otp"]["max_submit_attempts"] == 2 and cfg["captcha"]["expected_length"] == 5, str(cfg["nsdl_otp"]))
|
|
for bad_env, label in (({"NSDL_OTP_WAIT_SECONDS": "601"}, "wait > validity"),
|
|
({"NSDL_OTP_MAX_SUBMIT_ATTEMPTS": "5"}, "5 submits"),
|
|
({"NSDL_OTP_FRAPPE_TIME_ZONE": "Mars/Base"}, "bad time zone"),
|
|
({"NSDL_CAPTCHA_MIN_AGREEMENT": "2"}, "bad agreement")):
|
|
try:
|
|
build_nsdl_login_cfg(acct, environ=bad_env)
|
|
check(f"rejects {label}", False)
|
|
except NsdlLoginConfigError:
|
|
check(f"rejects {label}", True)
|
|
check("empty captcha length env -> any length", build_nsdl_login_cfg(acct, environ={"NSDL_CAPTCHA_EXPECTED_LENGTH": ""})["captcha"]["expected_length"] is None)
|
|
try:
|
|
FrappeOtpMailbox.from_config({"nsdl_otp": cfg["nsdl_otp"], "NSDL_USERNAME": "u"})
|
|
check("missing OTP_* keys reported", False)
|
|
except OtpMailboxError as exc:
|
|
check("missing OTP_* keys reported", "OTP_FRAPPE_URL" in str(exc))
|
|
|
|
print()
|
|
print("FAILED:" if failures else "ALL PASSED", failures or "")
|
|
sys.exit(1 if failures else 0)
|