2727from urllib .parse import urljoin , urlparse
2828
2929KINDS = {"" : "Not signed in" , "header" : "Header" , "cookie" : "Session cookie" ,
30- "form" : "Form login" }
30+ "form" : "Form login" , "recorded" : "Recorded in a browser" }
31+ # The kinds a person can type into the site's form. A recorded sign in only
32+ # comes from Record a session, so the form can keep it but never make one.
33+ FORM_KINDS = {k : v for k , v in KINDS .items () if k != "recorded" }
3134_HEADER_NAME = re .compile (r"^[A-Za-z0-9!#$%&'*+.^_`|~-]{1,100}$" )
3235_FIELD_NAME = re .compile (r"^[A-Za-z0-9_.\[\]:-]{1,100}$" )
3336
@@ -91,6 +94,16 @@ def describe(asset, slot: str = "a") -> dict:
9194 out .update (login_url = data .get ("login_url" , "" ), username = data .get ("username" , "" ),
9295 username_field = data .get ("username_field" , "" ),
9396 password_field = data .get ("password_field" , "" ))
97+ elif data ["kind" ] == "recorded" :
98+ from app .core import session_recorder
99+
100+ steps = data .get ("steps" ) or []
101+ out .update (cookie_names = [p .split ("=" , 1 )[0 ].strip ()
102+ for p in data .get ("cookie" , "" ).split (";" ) if "=" in p ],
103+ recorded_at = data .get ("recorded_at" , "" ),
104+ steps = session_recorder .public_steps (steps ),
105+ replayable = bool (steps ) and not any (s .get ("human" ) for s in steps ),
106+ has_totp = bool (data .get ("totp_secret" )))
94107 return out
95108
96109
@@ -124,7 +137,13 @@ def save(asset, kind: str, values: dict, actor: str, parent_domain: str = "",
124137 return
125138 old = load (asset , slot ) if getattr (asset , kind_attr ) == kind else {}
126139 data = {}
127- if kind == "header" :
140+ if kind == "recorded" :
141+ # Only Record a session makes one. The site's form keeps it as it is,
142+ # so saving the form for another reason does not lose the recording.
143+ if not old :
144+ raise ValueError ("Use Record a session to sign in through a browser." )
145+ data = {k : v for k , v in old .items () if not k .startswith ("_" ) and k != "kind" }
146+ elif kind == "header" :
128147 name = _clean_line (values .get ("header_name" ) or old .get ("header_name" ) or "" , "Header name" , 100 )
129148 if not _HEADER_NAME .fullmatch (name ):
130149 raise ValueError ("Give the header's name, such as Authorization." )
@@ -178,6 +197,61 @@ def save(asset, kind: str, values: dict, actor: str, parent_domain: str = "",
178197 setattr (asset , at_attr , utcnow ())
179198
180199
200+ def save_recorded (asset , recording : dict , actor : str , start_url : str ,
201+ slot : str = "a" ) -> None :
202+ """Keep what Record a session captured as the site's sign in."""
203+ from app .core import session_recorder
204+ from app .core .models import utcnow
205+
206+ kind_attr , data_attr , by_attr , at_attr = _SLOTS [slot ]
207+ steps = recording .get ("steps" ) or []
208+ data = {"cookie" : session_recorder .cookie_header (recording .get ("cookies" ) or []),
209+ "user_agent" : str (recording .get ("user_agent" ) or "" )[:500 ],
210+ "steps" : steps , "start_url" : start_url [:1000 ],
211+ "final_url" : str (recording .get ("final_url" ) or "" )[:1000 ],
212+ "recorded_at" : utcnow ().isoformat ()}
213+ if recording .get ("totp_secret" ):
214+ data ["totp_secret" ] = recording ["totp_secret" ]
215+ prior = load (asset , slot )
216+ if prior .get ("intent" ):
217+ data ["intent" ] = prior ["intent" ]
218+ setattr (asset , kind_attr , "recorded" )
219+ setattr (asset , data_attr , _encrypt (data ))
220+ setattr (asset , by_attr , (actor or "" )[:120 ])
221+ setattr (asset , at_attr , utcnow ())
222+
223+
224+ def check_live (data : dict , cookie : str , allow_private : bool = False ) -> tuple :
225+ """Whether a recorded session still gets past the site's front door.
226+ -> (ok, a sentence). A bot check answering, or the sign in page coming
227+ back, means the scan would see only that, so it is said plainly."""
228+ from app .core import external_probe
229+
230+ # The page the recording ended on, when it is the site's own: a sign in
231+ # start page can show its form whether or not the session is good.
232+ final = data .get ("final_url" ) or ""
233+ host = (data .get ("_host" ) or "" ).lower ()
234+ final_host = (urlparse (final ).hostname or "" ).lower ()
235+ url = (final if final_host and host and (final_host == host or final_host .endswith ("." + host ))
236+ else data .get ("start_url" ) or (f"https://{ host } /" if host else "" ))
237+ if not url :
238+ return True , ""
239+ headers = {"Cookie" : cookie }
240+ if data .get ("user_agent" ):
241+ headers ["User-Agent" ] = data ["user_agent" ]
242+ got = external_probe .fetch (url , headers = headers , allow_private = allow_private )
243+ if got .get ("error" ) and not got .get ("status" ):
244+ return False , f"the site did not answer: { str (got ['error' ])[:100 ]} "
245+ body = got .get ("body" ) or ""
246+ check = external_probe .bot_check (body , got .get ("status" ) or 0 )
247+ if check :
248+ return False , (f"{ check } 's bot check answers instead of the site, so the recorded "
249+ "pass has expired or is tied to something the scan cannot repeat" )
250+ if _login_form (body ) is not None and (data .get ("final_url" ) or "" ) != (got .get ("final_url" ) or "" ):
251+ return False , "the sign in page comes back, so the recorded session has ended"
252+ return True , ""
253+
254+
181255# ------------------------------------------------------------------ secrets
182256
183257def secrets_of (data : dict , session_cookie : str = "" ) -> list :
@@ -197,6 +271,11 @@ def secrets_of(data: dict, session_cookie: str = "") -> list:
197271 out .append (value )
198272 if data .get ("password" ):
199273 out .append (data ["password" ])
274+ for step in data .get ("steps" ) or []:
275+ if step .get ("secret" ) and step .get ("text" ):
276+ out .append (step ["text" ])
277+ if data .get ("totp_secret" ):
278+ out .append (data ["totp_secret" ])
200279 if data .get ("kind" ) == "header" and data .get ("header_value" , "" ).lower ().startswith ("basic " ):
201280 try :
202281 pair = base64 .b64decode (data ["header_value" ].split (" " , 1 )[1 ]).decode ()
@@ -387,9 +466,37 @@ def session_for(data: dict, allow_private: bool = False) -> dict:
387466 got = login (data , allow_private )
388467 out ["ok" ], out ["cookie" ] = got ["ok" ], got ["cookie" ]
389468 out ["detail" ] = got ["detail" ]
469+ elif kind == "recorded" :
470+ out .update (_recorded_session (data , allow_private ))
390471 return out
391472
392473
474+ def _recorded_session (data : dict , allow_private : bool ) -> dict :
475+ """A recorded sign in for a scan: the steps replayed when they can be,
476+ else the cookies the recording kept, and either way checked against the
477+ site before the engines lean on it."""
478+ from app .core import external_probe , session_recorder
479+
480+ cookie , how = data .get ("cookie" , "" ), "the recorded session"
481+ steps = data .get ("steps" ) or []
482+ if steps and not any (s .get ("human" ) for s in steps ):
483+ from app .core import settings_store
484+ from app .core .db import session_scope
485+
486+ with session_scope () as db :
487+ domains = settings_store .get (db , "web.company_domains" ) or []
488+ got = session_recorder .replay (data , allow_private , external_probe .current_proxy (),
489+ domains )
490+ if got ["ok" ]:
491+ cookie , how = got ["cookie" ], "signed in again by replaying the recorded steps"
492+ else :
493+ how = f"the recorded session (replaying the steps failed: { got ['detail' ]} )"
494+ headers = [("User-Agent" , data ["user_agent" ])] if data .get ("user_agent" ) else []
495+ ok , why = check_live (data , cookie , allow_private )
496+ return {"cookie" : cookie , "headers" : headers , "ok" : ok ,
497+ "detail" : how if ok else f"{ why } . Record the session again from the site page" }
498+
499+
393500def test_signin (slots : dict , allow_private : bool = False ) -> list :
394501 """Sign in now with each configured account, so an administrator can
395502 confirm it works before a scan leans on it. slots is {"a": data, "b":
@@ -412,7 +519,8 @@ def test_signin(slots: dict, allow_private: bool = False) -> list:
412519 if not kind :
413520 out .append (row )
414521 continue
415- if kind == "form" :
522+ if kind in ("form" , "recorded" ):
523+ row ["live" ] = True
416524 sess = session_for (data , allow_private )
417525 row ["ok" ], row ["detail" ] = sess ["ok" ], sess ["detail" ]
418526 else :
0 commit comments