aws_extender.py 文件源码

def check_timestamp(self, bucket_url, bucket_type, timestamp):
        """Check timestamps of signed URLs."""
        timestamp_raw = timestamp
        offsets = []
        mark_request = False
        start = 0

        try:
            if bucket_type != 'Azure':
                now = int(time.time())
                diff = (int(timestamp) - now) / 3600
            else:
                timestamp = unquote(timestamp)
                timestamp = datetime.strptime(timestamp, '%Y-%m-%dT%H:%M:%S%fZ')
                diff = int((timestamp - datetime.now()).total_seconds()) / 3600
        except ValueError:
            return

        if diff > 24:
            start = self.helpers.indexOf(self.response,
                                         timestamp_raw, True, 0, self.response_len)
            if start < 0:
                start = self.helpers.indexOf(self.request,
                                             timestamp_raw, True, 0, self.request_len)
                mark_request = True
            self.offset[0] = start
            self.offset[1] = start + len(timestamp_raw)
            offsets.append(self.offset)
            if mark_request:
                markers = [self.callbacks.applyMarkers(self.request_response, offsets, None)]
            else:
                markers = [self.callbacks.applyMarkers(self.request_response, None, offsets)]
            issue_name = '%s Signed URL Excessive Expiration Time' % bucket_type
            issue_level = 'Information'
            issue_detail = '''The following %s signed URL was found to be valid for more than
                24 hours (expires in %sh):<br><li>%s</li>''' % (bucket_type, diff, bucket_url)
            self.scan_issues.append(
                ScanIssue(self.request_response.getHttpService(),
                          self.current_url, markers, issue_name, issue_level, issue_detail)
            )