diff --git a/scripts/root_custody_finish.py b/scripts/root_custody_finish.py index fbd6cee..e832ac6 100644 --- a/scripts/root_custody_finish.py +++ b/scripts/root_custody_finish.py @@ -85,16 +85,24 @@ def inspect_share(lane, operator, inventory=None, network=None, progress=lambda expired=None if expiry: try: - parsed=datetime.fromisoformat(expiry).replace(tzinfo=timezone.utc) - expired=parsed<=datetime.now(timezone.utc) + parsed=datetime.fromisoformat(expiry) + if parsed.tzinfo is not None: + expired=parsed.astimezone(timezone.utc)<=datetime.now(timezone.utc) except (TypeError,ValueError):pass result.update(current_share_label_matches=share.get('label')==LABEL, current_share_owner_matches=share.get('uid_owner')=='Backup', current_share_path_matches=share.get('path')==FOLDER, current_share_public_link=share.get('share_type') in (3,'3'), current_share_create_only=share.get('permissions') in (4,'4'), - current_share_password_indicator=bool(share.get('password')), - current_share_expiry_present=bool(expiry),current_share_expired=expired) + current_share_password_indicator=bool(share['password']) if 'password' in share else None, + current_share_password_indicator_known='password' in share, + current_share_expiry_present=bool(expiry),current_share_expired=expired, + current_share_expiry_clock_basis_known=not bool(expiry) or expired is not None) + if result['current_share_password_indicator'] is True or expired is True: + result['public_route_probes_skipped']=True + result['public_route_skip_reason']='password_protected' if result['current_share_password_indicator'] is True else 'expired' + return result + result['public_route_probes_skipped']=False progress('share_route_probes') body=b'' headers={'Depth':'0','Content-Type':'application/xml','X-Requested-With':'XMLHttpRequest'} diff --git a/tests/test_root_custody_finish.py b/tests/test_root_custody_finish.py index 894d752..0760a2a 100644 --- a/tests/test_root_custody_finish.py +++ b/tests/test_root_custody_finish.py @@ -269,9 +269,10 @@ class ShareInspectionTests(unittest.TestCase): result=finish.inspect_share(lane,operator,inventory=lambda *args:inventory,network=lambda *args:(401,b'')) self.assertFalse(result['current_token_unique']);self.assertNotIn('current_share_create_only',result) def test_password_and_expiry_indicators_do_not_emit_values(self): - lane,operator,shares=self.fixture();shares[0].update(password='PRIVATE-HASH',expiration='2000-01-01 00:00:00') + lane,operator,shares=self.fixture();shares[0].update(password='PRIVATE-HASH',expiration='2000-01-01T00:00:00+02:00') result=finish.inspect_share(lane,operator,inventory=lambda *args:shares,network=lambda *args:(401,b'')) self.assertTrue(result['current_share_password_indicator']);self.assertTrue(result['current_share_expired']) + self.assertTrue(result['public_route_probes_skipped']) self.assertNotIn('PRIVATE-HASH',json.dumps(result)) def test_main_diagnostic_cannot_call_upload(self): case=MainIntegrationTests('test_full_attended_main_success_real_crypto_fake_owner_and_provider') @@ -288,3 +289,19 @@ class ShareInspectionTests(unittest.TestCase): self.assertEqual(receipt['scope'],'read-only-current-governed-share-inspection') self.assertTrue(all(x[1]=='PROPFIND' for x in calls)) finally:case.tearDown() + + def test_naive_expiry_and_missing_password_remain_unknown(self): + lane,operator,shares=self.fixture();shares[0].pop('password') + shares[0]['expiration']='2000-01-01 00:00:00' + result=finish.inspect_share(lane,operator,inventory=lambda *args:shares,network=lambda *args:(403,b'')) + self.assertIsNone(result['current_share_expired']) + self.assertFalse(result['current_share_expiry_clock_basis_known']) + self.assertIsNone(result['current_share_password_indicator']) + self.assertFalse(result['current_share_password_indicator_known']) + def test_clearly_protected_or_expired_share_skips_public_auth_retries(self): + for protection in ('password','expiration'): + lane,operator,shares=self.fixture() + shares[0][protection]='PRIVATE' if protection=='password' else '2000-01-01T00:00:00-02:00' + network=unittest.mock.Mock(side_effect=AssertionError('NO-AUTH-RETRY')) + result=finish.inspect_share(lane,operator,inventory=lambda *args:shares,network=network) + self.assertTrue(result['public_route_probes_skipped']);network.assert_not_called()