import re import tarfile import json from pathlib import Path import yaml import pydantic from rest_framework import serializers from labs.lab_manager_v2.lab_guide import ( LabGuidePart, LabGuideMarkdown, LabGuideQuiz, LabGuideQuizContent, ) from labs.lab_manager_v2.lab_spec import ( LabSpec, BootstrapShellScriptStep, CloudStackRequirement, TiDBEnvironmentType, ) def is_subpath(child: str | Path, parent: str | Path) -> bool: try: Path(child).resolve(strict=False).relative_to( Path(parent).resolve(strict=False) ) return True except ValueError: return False MANIFEST_FILE = "manifest.yaml" AWS_CLOUDFORMATION_FILE = "cloudformation.json" ALICLOUD_ROS_FILE = "alicloud_ros.json" GUIDES_DIR = "guides" LAB_CODE_PATTERN = r"^[a-z0-9_]+$" class LabSpecSerializer(serializers.Serializer): """Validate the lab spec file. The uploaded file must be a ``.tar.gz`` archive containing: - ``manifest.yaml`` - ``guides/`` - ``cloudformation.json`` or ``alicloud_ros.json`` """ file = serializers.FileField(required=True) def validate_file(self, file): if not file.name.endswith(".tar.gz"): raise serializers.ValidationError("File must be a tar.gz file") if file.size > (2 << 20): raise serializers.ValidationError("File must be less than 2MB") with tarfile.open(fileobj=file.file, mode="r:gz") as tar: members = tar.getmembers() top_level_dir_name = self._ensure_only_one_top_dir(members) lab_code = top_level_dir_name self._validate_lab_code(lab_code) lab_spec = self._validate_manifest(lab_code, tar) self._validate_aws_cloudformation(lab_spec, tar) self._validate_alicloud_ros(lab_spec, tar) self._validate_guides(lab_code, tar) return file def _validate_lab_code(self, lab_code: str) -> None: if not re.match(LAB_CODE_PATTERN, lab_code): raise serializers.ValidationError( "The lab code must match the pattern: " + LAB_CODE_PATTERN ) def _validate_aws_cloudformation( self, lab_spec: LabSpec, tar: tarfile.TarFile ) -> None: if not lab_spec.regions.aws: # This lab does not specify AWS regions, so it does not need a cloudformation file return cloudformation_file = Path(lab_spec.lab_code) / AWS_CLOUDFORMATION_FILE if str(cloudformation_file) not in [member.name for member in tar.getmembers()]: raise serializers.ValidationError( f"The cloudformation file {cloudformation_file} does not exist in the archive" ) try: content = tar.extractfile(str(cloudformation_file)).read() if len(content) > 51200: raise serializers.ValidationError( f"The cloudformation file {cloudformation_file} is too large, size: {len(content)} bytes, it must be less than 51200 bytes" ) json.loads(content) except json.JSONDecodeError as e: raise serializers.ValidationError( f"The cloudformation file {cloudformation_file} is not a valid JSON file: {e}" ) except Exception as e: raise serializers.ValidationError( f"Failed to parse the cloudformation file {cloudformation_file}: {e}" ) def _validate_alicloud_ros(self, lab_spec: LabSpec, tar: tarfile.TarFile) -> None: if not lab_spec.regions.alicloud: # This lab does not specify Alicloud regions, so it does not need a ros file return ros_file = Path(lab_spec.lab_code) / ALICLOUD_ROS_FILE if str(ros_file) not in [member.name for member in tar.getmembers()]: raise serializers.ValidationError( f"The ros file {ros_file} does not exist in the archive" ) try: content = tar.extractfile(str(ros_file)).read() if len(content) > 524288: raise serializers.ValidationError( f"The ros file {ros_file} is too large, size: {len(content)} bytes, it must be less than 524288 bytes" ) json.loads(content) except json.JSONDecodeError as e: raise serializers.ValidationError( f"The ros file {ros_file} is not a valid JSON file: {e}" ) except Exception as e: raise serializers.ValidationError( f"Failed to parse the ros file {ros_file}: {e}" ) def _validate_manifest(self, lab_code: str, tar: tarfile.TarFile) -> LabSpec: try: manifest_file = tar.extractfile(f"{lab_code}/{MANIFEST_FILE}") except KeyError: raise serializers.ValidationError( f"The archive must contain a {MANIFEST_FILE} file" ) try: manifest = yaml.safe_load(manifest_file.read()) lab_spec = LabSpec.model_validate(manifest) except pydantic.ValidationError as e: raise serializers.ValidationError( f"The {MANIFEST_FILE} file is invalid: {e}" ) except Exception as e: raise serializers.ValidationError( f"Failed to parse the {MANIFEST_FILE} file: {e}" ) if lab_spec.lab_code != lab_code: raise serializers.ValidationError( "The lab code in the manifest file does not match the lab code in the file name: " f"{lab_spec.lab_code} != {lab_code}" ) top_level_dir = Path(lab_code) member_names = [member.name for member in tar.getmembers()] for bootstrap_step in lab_spec.bootstrap: if isinstance(bootstrap_step, BootstrapShellScriptStep): if bootstrap_step.script: script_path = top_level_dir / bootstrap_step.script if not is_subpath(script_path, top_level_dir): raise serializers.ValidationError( f"The script file {script_path} is not in the top-level directory: {top_level_dir}" ) if str(script_path) not in member_names: raise serializers.ValidationError( f"The script file {script_path} does not exist in the archive" ) cloud_stack_requirement_exists = False for requirement in lab_spec.requirements: if isinstance(requirement, CloudStackRequirement): cloud_stack_requirement_exists = True if not cloud_stack_requirement_exists: raise serializers.ValidationError( "The lab spec must have a CloudStack requirement" ) return lab_spec ALLOWED_LANGUAGES = ["en", "ja", "zh", "ko"] def _validate_guides(self, lab_code: str, tar: tarfile.TarFile) -> None: members = tar.getmembers() member_names = [member.name for member in members] lab_guides_dir = Path(lab_code) / GUIDES_DIR actual_guides_languages = [] for member in members: if member.isdir() and member.name.startswith(str(lab_guides_dir)): if member.name == str(lab_guides_dir): continue guide_language = member.name.split("/")[-1] if guide_language not in self.ALLOWED_LANGUAGES: raise serializers.ValidationError( f"The guide language {guide_language} is not allowed" ) actual_guides_languages.append(guide_language) for language in actual_guides_languages: catalog_file = lab_guides_dir / language / "catalog.yaml" try: catalog_content = yaml.safe_load( tar.extractfile(str(catalog_file)).read() ) parts = pydantic.TypeAdapter(list[LabGuidePart]).validate_python( catalog_content ) except pydantic.ValidationError as e: raise serializers.ValidationError( f"The catalog file {catalog_file} is invalid: {e}" ) except Exception as e: raise serializers.ValidationError( f"Failed to parse the catalog file {catalog_file}: {e}" ) for p in parts: if isinstance(p, LabGuideMarkdown) or isinstance(p, LabGuideQuiz): resource_file = (lab_guides_dir / language) / p.file else: continue base_dir = lab_guides_dir / language if not is_subpath(resource_file, base_dir): raise serializers.ValidationError( f"The guide file {resource_file} is not in the guides directory: {base_dir}" ) if str(resource_file) not in member_names: raise serializers.ValidationError( f"The guide file {resource_file} does not exist in the archive" ) if isinstance(p, LabGuideQuiz): try: LabGuideQuizContent.model_validate( yaml.safe_load(tar.extractfile(str(resource_file)).read()) ) except pydantic.ValidationError as e: raise serializers.ValidationError( f"The quiz file {resource_file} is invalid: {e}" ) def _ensure_only_one_top_dir(self, members: list[tarfile.TarInfo]) -> str: top_level_entries = set() for member in members: # Filter out metadata files like PaxHeaders. if ( member.isdev() or member.isfifo() or member.islnk() or member.issym() or member.ischr() ): continue # In a tar archive, a path can be split by '/'. The first part is the top-level entry. top_level_entry = member.name.split("/")[0] if ( top_level_entry ): # Exclude empty strings that might result from paths like './' top_level_entries.add(top_level_entry) if not top_level_entries: raise serializers.ValidationError( "The archive is empty or contains no files." ) if len(top_level_entries) > 1: raise serializers.ValidationError( f"The archive must contain only one top-level directory. " f"Found multiple entries: {sorted(list(top_level_entries))}" ) # At this point, there is only one top-level entry. Now, ensure it's a directory. # We can do this by checking if there are any files at the root level. # A file at the root level would not have a '/' in its name (unless it starts with './'). for member in members: if member.isfile() and "/" not in member.name.strip("/"): raise serializers.ValidationError( "The archive must contain only one top-level directory. " f"Found a file at the root level: {member.name}" ) return top_level_entries.pop()