/actual/resource/path"
#
# Here we need to extract the "/actual/resource/path" part.
segments_to_drop = len(self.payload_data.pattern_path_segments) + 1
resource_path = request_info.url.path_segments[segments_to_drop:]
if resource_path == ():
return resource_blocked_response
elif resource_path[0] == 'static':
return self._make_file_resource_response(resource_path[1:])
elif resource_path[0] == 'api':
# TODO: implement Haketilo APIs
return resource_blocked_response
else:
return resource_blocked_response
resource_blocked_response = http_messages.ProducedResponse(
403,
[(b'Content-Type', b'text/plain; charset=utf-8')],
_('api.resource_not_enabled_for_access').encode()
)
@dc.dataclass(frozen=True)
class BlockedResponsePolicy(base.Policy):
"""...."""
process_request: t.ClassVar[bool] = True
priority: t.ClassVar[base.PolicyPriority] = base.PolicyPriority._THREE
def consume_request(self, request_info: http_messages.RequestInfo) \
-> http_messages.ProducedResponse:
"""...."""
return resource_blocked_response
@dc.dataclass(frozen=True, unsafe_hash=True) # type: ignore[misc]
class PayloadResourcePolicyFactory(PayloadAwarePolicyFactory):
"""...."""
def make_policy(self, haketilo_state: state.HaketiloState) \
-> t.Union[PayloadResourcePolicy, BlockedResponsePolicy]:
"""...."""
try:
payload_data = self.payload_ref.get_data()
except state.MissingItemError:
return BlockedResponsePolicy()
if not payload_data.explicitly_enabled and \
haketilo_state.get_settings().mapping_use_mode != \
state.MappingUseMode.AUTO:
return BlockedResponsePolicy()
return PayloadResourcePolicy(payload_data)
lass='msg-avail'>...Ludovic Courtès |
2016-11-23 | marionette: Delay synchronization with the host's REPL.... | Ludovic Courtès |
2016-11-22 | marionette: Avoid use of SIGALRM for timeouts.... | Ludovic Courtès |
2016-05-04 | Add (gnu tests) and (gnu build marionette).... | Ludovic Courtès |