1+ #!/usr/bin/env python3
2+
3+ from __future__ import annotations
4+
5+ import json
6+ import os
7+ import re
8+ import sys
9+ import time
10+ from pathlib import Path
11+ from typing import Iterable
12+ from urllib .error import URLError , HTTPError
13+ from urllib .request import Request , urlopen
14+
15+
16+ REPO_ROOT = Path (__file__ ).resolve ().parents [1 ]
17+ DEFAULT_API_URL = "https://pim.api.stackit.cloud/v1/skus"
18+ API_TIMEOUT_SECONDS = 45
19+ API_RETRIES = 3
20+
21+
22+ def fetch_json (url : str ) -> dict :
23+ last_error : Exception | None = None
24+ headers = {"User-Agent" : "stackit-landing-zone-flavor-validator/1.0" }
25+
26+ for attempt in range (1 , API_RETRIES + 1 ):
27+ try :
28+ request = Request (url , headers = headers )
29+ with urlopen (request , timeout = API_TIMEOUT_SECONDS ) as response :
30+ if response .status != 200 :
31+ raise RuntimeError (f"Unexpected HTTP status { response .status } { url } )
32+ payload = response .read ().decode ("utf-8" )
33+ return json .loads (payload )
34+ except (HTTPError , URLError , TimeoutError , json .JSONDecodeError , RuntimeError ) as error :
35+ last_error = error
36+ if attempt < API_RETRIES :
37+ time .sleep (attempt )
38+
39+ raise RuntimeError (f"Failed to fetch STACKIT SKUs from { url } { last_error } )
40+
41+
42+ def extract_live_flavors (payload : dict ) -> tuple [set [str ], set [str ]]:
43+ server_flavors : set [str ] = set ()
44+ git_flavors : set [str ] = set ()
45+
46+ if "services" in payload :
47+ items = payload .get ("services" , [])
48+ for item in items :
49+ if not isinstance (item , dict ):
50+ continue
51+
52+ product = str (item .get ("product" ) or "" )
53+ deprecated = str (item .get ("deprecated" ) or "" )
54+ if deprecated .lower () == "yes" :
55+ continue
56+
57+ attributes = item .get ("attributes" )
58+ if not isinstance (attributes , dict ):
59+ attributes = {}
60+
61+ if product == "Server" :
62+ flavor = attributes .get ("flavor" )
63+ if isinstance (flavor , str ) and flavor .strip ():
64+ server_flavors .add (flavor .strip ())
65+
66+ if product == "Git" :
67+ name = str (item .get ("name" ) or "" )
68+ match = re .match (r"^Git-(\d+)-" , name )
69+ if match :
70+ git_flavors .add (f"git-{ match .group (1 )} )
71+
72+ elif "data" in payload :
73+ items = payload .get ("data" , [])
74+ for item in items :
75+ if not isinstance (item , dict ):
76+ continue
77+
78+ product = str (item .get ("productName" ) or "" )
79+ deprecated = str (item .get ("deprecated" ) or "" )
80+ if deprecated .lower () == "yes" :
81+ continue
82+
83+ attributes = item .get ("productSpecificAttributes" )
84+ if not isinstance (attributes , dict ):
85+ attributes = {}
86+
87+ if product == "Server" :
88+ flavor = attributes .get ("flavor" )
89+ if isinstance (flavor , str ) and flavor .strip ():
90+ server_flavors .add (flavor .strip ())
91+
92+ if product == "Git" :
93+ name = str (item .get ("name" ) or "" )
94+ match = re .match (r"^Git-(\d+)-" , name )
95+ if match :
96+ git_flavors .add (f"git-{ match .group (1 )} )
97+ else :
98+ raise RuntimeError ("Unsupported SKU API response format: expected 'services' or 'data'." )
99+
100+ if not server_flavors :
101+ raise RuntimeError ("No live server flavors found in SKU API response." )
102+ if not git_flavors :
103+ raise RuntimeError ("No live git flavors found in SKU API response." )
104+
105+ return server_flavors , git_flavors
106+
107+
108+ def iter_tf_files (root : Path ) -> Iterable [Path ]:
109+ for path in root .rglob ("*" ):
110+ if not path .is_file ():
111+ continue
112+ if path .suffix not in ".tf" , ".tfvars" }:
113+ continue
114+ if ".terraform" in path .parts :
115+ continue
116+ yield path
117+
118+
119+ def collect_used_flavors (root : Path ) -> tuple [list [tuple [str , str ]], list [tuple [str , str ]]]:
120+ used_server : list [tuple [str , str ]] = []
121+ used_git : list [tuple [str , str ]] = []
122+
123+ assignment_server_re = re .compile (r"\b(?:machine_type|firewall_flavor)\s*=\s*\"([^\"]+)\"" )
124+ assignment_git_re = re .compile (r"\bgit_flavor\s*=\s*\"([^\"]+)\"" )
125+ block_start_re = re .compile (r'^\s*variable\s+"(firewall_flavor|git_flavor)"\s*{\s*$' )
126+ default_re = re .compile (r'\bdefault\s*=\s*"([^\"]+)"' )
127+
128+ for path in iter_tf_files (root ):
129+ rel_path = path .relative_to (root )
130+ in_var_block : str | None = None
131+
132+ with path .open ("r" , encoding = "utf-8" ) as handle :
133+ for line_number , line in enumerate (handle , start = 1 ):
134+ code = line .split ("#" , 1 )[0 ].split ("//" , 1 )[0 ]
135+
136+ start_match = block_start_re .match (code )
137+ if start_match :
138+ in_var_block = start_match .group (1 )
139+
140+ for match in assignment_server_re .finditer (code ):
141+ used_server .append ((match .group (1 ), f"{ rel_path } { line_number } ))
142+
143+ for match in assignment_git_re .finditer (code ):
144+ used_git .append ((match .group (1 ), f"{ rel_path } { line_number } ))
145+
146+ if in_var_block :
147+ default_match = default_re .search (code )
148+ if default_match :
149+ value = default_match .group (1 )
150+ if in_var_block == "firewall_flavor" :
151+ used_server .append ((value , f"{ rel_path } { line_number } ))
152+ elif in_var_block == "git_flavor" :
153+ used_git .append ((value , f"{ rel_path } { line_number } ))
154+
155+ if in_var_block and "}" in code :
156+ in_var_block = None
157+
158+ return used_server , used_git
159+
160+
161+ def validate (used : list [tuple [str , str ]], allowed : set [str ], kind : str ) -> list [str ]:
162+ errors : list [str ] = []
163+ for value , location in used :
164+ if value not in allowed :
165+ errors .append (
166+ f"{ kind } { value } { location }
167+ )
168+ return errors
169+
170+
171+ def main () -> int :
172+ api_url = os .environ .get ("STACKIT_PIM_SKUS_URL" , DEFAULT_API_URL )
173+
174+ try :
175+ payload = fetch_json (api_url )
176+ allowed_server , allowed_git = extract_live_flavors (payload )
177+ used_server , used_git = collect_used_flavors (REPO_ROOT )
178+
179+ errors = []
180+ errors .extend (validate (used_server , allowed_server , "server" ))
181+ errors .extend (validate (used_git , allowed_git , "git" ))
182+
183+ if errors :
184+ print ("Live flavor validation failed:" )
185+ for error in errors :
186+ print (f"- { error } )
187+
188+ print ("\n Allowed server flavor count:" , len (allowed_server ))
189+ print ("Allowed git flavor count:" , len (allowed_git ))
190+ return 1
191+
192+ print ("Live flavor validation succeeded." )
193+ print ("Validated server flavors:" , len (used_server ))
194+ print ("Validated git flavors:" , len (used_git ))
195+ print ("Allowed server flavors:" , len (allowed_server ))
196+ print ("Allowed git flavors:" , len (allowed_git ))
197+ return 0
198+ except Exception as error :
199+ print (f"Live flavor validation error: { error } , file = sys .stderr )
200+ return 2
201+
202+
203+ if __name__ == "__main__" :
204+ raise SystemExit (main ())
0 commit comments