-
Notifications
You must be signed in to change notification settings - Fork 3
/
example_aws_iam_authenticator.py
64 lines (62 loc) · 2.05 KB
/
example_aws_iam_authenticator.py
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
from fluentm.entities import (
Boundary,
Process,
DataFlow,
Exec,
HTTP,
SIGV4,
TLS,
SIGV4,
Stdout,
Internal,
)
from fluentm.renderer import report
scenes = {
# Example using variables, which is fine for small things but gets hard with longer flows
"kubectl gets pre-signed URL": [
DataFlow(
Process("kubectl").inBoundary("User Machine"),
Process("aws-cli").inBoundary("User Machine"),
Exec("Exec aws-cli get-token"),
),
DataFlow(
Process("aws-cli"),
Process("aws-cli"),
Internal("Sign URL using AWS IAM credentials"),
),
DataFlow(Process("aws-cli"), Process("kubectl"), Stdout("STS pre-signed URL")),
],
"API traffic": [
DataFlow(
Process("kubectl"),
Process("Kubernetes API").inBoundary(
Boundary("EKS Cluster").inBoundary("AWS Cloud")
),
TLS(HTTP("pre-signed URL as Bearer Token HTTP Header")),
),
DataFlow(
Process("Kubernetes API"),
Process("aws-iam-authenticator").inBoundary("EKS Cluster"),
TLS(HTTP("TokenReview request with pre-signed URL")),
),
DataFlow(
Process("aws-iam-authenticator"),
Process("AWS STS").inBoundary("AWS Cloud"),
TLS(HTTP(SIGV4("sts:GetCallerIdentity request"))),
response=TLS(HTTP("sts:GetCallerIdentity response")),
),
DataFlow(
Process("aws-iam-authenticator"),
Process("Kubernetes API").inBoundary("EKS Cluster"),
TLS(HTTP("TokenReview response with username")),
),
DataFlow(
Process("aws-iam-authenticator"),
Process("Kubernetes API").inBoundary("EKS Cluster"),
TLS(HTTP(("Async Watch mapped aws-auth ConfigMap "))),
# response=TLS(HTTP("Config Map username mappings")),
),
],
}
if __name__ == "__main__":
report(scenes, outputDir="aws-iam-authenticator", dfdLabels=True)