1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
|
"""Handles incoming signer requests, invokes methods, returns responses."""
import json
from typing import Any
from urllib.parse import unquote
from moto.core.responses import BaseResponse
from .models import SignerBackend, signer_backends
class SignerResponse(BaseResponse):
def __init__(self) -> None:
super().__init__(service_name="signer")
@property
def signer_backend(self) -> SignerBackend:
"""Return backend instance specific for this region."""
return signer_backends[self.current_account][self.region]
def cancel_signing_profile(self) -> str:
profile_name = self.path.split("/")[-1]
self.signer_backend.cancel_signing_profile(profile_name=profile_name)
return "{}"
def get_signing_profile(self) -> str:
profile_name = self.path.split("/")[-1]
profile = self.signer_backend.get_signing_profile(profile_name=profile_name)
return json.dumps(profile.to_dict())
def put_signing_profile(self) -> str:
params = json.loads(self.body)
profile_name = self.path.split("/")[-1]
signature_validity_period = params.get("signatureValidityPeriod")
platform_id = params.get("platformId")
tags = params.get("tags")
signing_material = params.get("signingMaterial")
profile = self.signer_backend.put_signing_profile(
profile_name=profile_name,
signature_validity_period=signature_validity_period,
platform_id=platform_id,
signing_material=signing_material,
tags=tags,
)
return json.dumps(profile.to_dict(full=False))
def list_signing_platforms(self) -> str:
platforms = self.signer_backend.list_signing_platforms()
return json.dumps({"platforms": platforms})
def list_tags_for_resource(self) -> str:
resource_arn = unquote(self.path.split("/tags/")[-1])
return json.dumps(
{"tags": self.signer_backend.list_tags_for_resource(resource_arn)}
)
def tag_resource(self) -> str:
resource_arn = unquote(self.path.split("/tags/")[-1])
tags = self._get_param("tags")
self.signer_backend.tag_resource(resource_arn, tags)
return "{}"
def untag_resource(self) -> str:
resource_arn = unquote(self.path.split("/tags/")[-1])
tag_keys = self.querystring.get("tagKeys")
self.signer_backend.untag_resource(resource_arn, tag_keys) # type: ignore
return "{}"
def tags(self, request: Any, full_url: str, headers: Any) -> str: # type: ignore[return]
self.setup_class(request, full_url, headers)
if request.method == "GET":
return self.list_tags_for_resource()
if request.method == "POST":
return self.tag_resource()
if request.method == "DELETE":
return self.untag_resource()
|
