|
| 1 | +import os |
| 2 | +from functools import cached_property |
| 3 | +from pathlib import Path |
| 4 | +from typing import Dict |
| 5 | + |
| 6 | +import attr |
| 7 | +import yaml |
| 8 | + |
| 9 | +from wafp.targets import ( |
| 10 | + BaseTarget, |
| 11 | + Language, |
| 12 | + Metadata, |
| 13 | + Package, |
| 14 | + SchemaSource, |
| 15 | + SchemaSourceType, |
| 16 | + Specification, |
| 17 | + SpecificationType, |
| 18 | +) |
| 19 | + |
| 20 | +KUBECONFIG_FILENAME = "admin.kubeconfig" |
| 21 | +KCP_DATA_PATH = Path(__file__).parent.absolute() / "kcp_data" / ".kcp" |
| 22 | + |
| 23 | + |
| 24 | +@attr.s |
| 25 | +class Default(BaseTarget): |
| 26 | + # TODO: |
| 27 | + # - proper cleanup of kcp_data folder |
| 28 | + # and / or |
| 29 | + # - move etcd dump to targets artifacts folder after run ends |
| 30 | + |
| 31 | + # port: int = attr.field(factory=lambda: 6443) |
| 32 | + fuzzer_skip_ssl_verify: bool = attr.ib(default=True) |
| 33 | + |
| 34 | + def get_base_url(self) -> str: |
| 35 | + # self.port does not work despite it was overrided in this child class, random came anyway |
| 36 | + # hardcode it for now |
| 37 | + return "https://0.0.0.0:6443" |
| 38 | + |
| 39 | + def get_schema_location(self) -> str: |
| 40 | + return f"{self.get_base_url()}/openapi/v2" |
| 41 | + |
| 42 | + def get_environment_variables(self): |
| 43 | + env = super().get_environment_variables() |
| 44 | + env.update({ |
| 45 | + "UID": f"{os.getuid()}", |
| 46 | + "GID": f"{os.getgid()}" |
| 47 | + }) |
| 48 | + print(env) |
| 49 | + return env |
| 50 | + |
| 51 | + def is_ready(self, line: bytes) -> bool: |
| 52 | + return b"Reconciling namespace root|default" in line |
| 53 | + |
| 54 | + def get_metadata(self) -> Metadata: |
| 55 | + return Metadata( |
| 56 | + language=Language.GO, |
| 57 | + framework=Package(name="kubernetes", version="1.23.5"), |
| 58 | + schema_source=SchemaSource( |
| 59 | + type=SchemaSourceType.GENERATED, library=Package(name="kubebuilder", version="unknown") |
| 60 | + ), |
| 61 | + validation_from_schema=True, |
| 62 | + specification=Specification(name=SpecificationType.OPENAPI, version="2.0"), |
| 63 | + ) |
| 64 | + |
| 65 | + def after_start(self, stdout: bytes, headers: Dict[str, str]) -> None: |
| 66 | + headers["Authorization"] = f"Bearer {self.auth_token}" |
| 67 | + |
| 68 | + @cached_property |
| 69 | + def auth_token(self) -> str: |
| 70 | + with open(KCP_DATA_PATH / KUBECONFIG_FILENAME, "r") as fd: |
| 71 | + kubeconfig = fd.read() |
| 72 | + kubeconfig = yaml.safe_load(kubeconfig) |
| 73 | + # TODO look up for kubeconfig type |
| 74 | + token: str = kubeconfig["users"][0]["user"]["token"] # type: ignore |
| 75 | + return token |
0 commit comments