-
Notifications
You must be signed in to change notification settings - Fork 121
Expand file tree
/
Copy pathreference.py
More file actions
1055 lines (882 loc) · 40.9 KB
/
reference.py
File metadata and controls
1055 lines (882 loc) · 40.9 KB
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
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
"""DependencyReference model -- core dependency representation and parsing."""
import re
import urllib.parse
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Dict, List, Optional
from ...utils.github_host import (
default_host,
is_artifactory_path,
is_azure_devops_hostname,
is_github_hostname,
is_supported_git_host,
parse_artifactory_path,
unsupported_host_error,
)
from ...utils.path_security import (
PathTraversalError,
ensure_path_within,
validate_path_segments,
)
from ..validation import InvalidVirtualPackageExtensionError
from .types import VirtualPackageType
@dataclass
class DependencyReference:
"""Represents a reference to an APM dependency."""
repo_url: str # e.g., "user/repo" for GitHub or "org/project/repo" for Azure DevOps
host: Optional[str] = (
None # Optional host (github.com, dev.azure.com, or enterprise host)
)
reference: Optional[str] = None # e.g., "main", "v1.0.0", "abc123"
alias: Optional[str] = None # Optional alias for the dependency
virtual_path: Optional[str] = (
None # Path for virtual packages (e.g., "prompts/file.prompt.md")
)
is_virtual: bool = (
False # True if this is a virtual package (individual file or collection)
)
# Azure DevOps specific fields (ADO uses org/project/repo structure)
ado_organization: Optional[str] = None # e.g., "dmeppiel-org"
ado_project: Optional[str] = None # e.g., "market-js-app"
ado_repo: Optional[str] = None # e.g., "compliance-rules"
# Local path dependency fields
is_local: bool = False # True if this is a local filesystem dependency
local_path: Optional[str] = (
None # Original local path string (e.g., "./packages/my-pkg")
)
artifactory_prefix: Optional[str] = (
None # e.g., "artifactory/github" (repo key path)
)
# Preserved verbatim when the user supplied an explicit ssh:// URL in apm.yml.
# Used by the downloader to clone with the exact URL (including any custom port)
# instead of the reconstructed https:// fallback URL.
original_ssh_url: Optional[str] = None
# Supported file extensions for virtual packages
VIRTUAL_FILE_EXTENSIONS = (
".prompt.md",
".instructions.md",
".chatmode.md",
".agent.md",
)
def is_artifactory(self) -> bool:
"""Check if this reference points to a JFrog Artifactory VCS repository."""
return self.artifactory_prefix is not None
def is_azure_devops(self) -> bool:
"""Check if this reference points to Azure DevOps."""
from ...utils.github_host import is_azure_devops_hostname
return self.host is not None and is_azure_devops_hostname(self.host)
@property
def virtual_type(self) -> "Optional[VirtualPackageType]":
"""Return the type of virtual package, or None if not virtual."""
if not self.is_virtual or not self.virtual_path:
return None
if any(self.virtual_path.endswith(ext) for ext in self.VIRTUAL_FILE_EXTENSIONS):
return VirtualPackageType.FILE
if "/collections/" in self.virtual_path or self.virtual_path.startswith(
"collections/"
):
return VirtualPackageType.COLLECTION
return VirtualPackageType.SUBDIRECTORY
def is_virtual_file(self) -> bool:
"""Check if this is a virtual file package (individual file)."""
return self.virtual_type == VirtualPackageType.FILE
def is_virtual_collection(self) -> bool:
"""Check if this is a virtual collection package."""
return self.virtual_type == VirtualPackageType.COLLECTION
def is_virtual_subdirectory(self) -> bool:
"""Check if this is a virtual subdirectory package (e.g., Claude Skill).
A subdirectory package is a virtual package that:
- Has a virtual_path that is NOT a file extension we recognize
- Is NOT a collection (doesn't have /collections/ in path)
- Is a directory path (likely containing SKILL.md or apm.yml)
Examples:
- ComposioHQ/awesome-claude-skills/brand-guidelines -> True
- owner/repo/prompts/file.prompt.md -> False (is_virtual_file)
- owner/repo/collections/name -> False (is_virtual_collection)
"""
return self.virtual_type == VirtualPackageType.SUBDIRECTORY
def get_virtual_package_name(self) -> str:
"""Generate a package name for this virtual package.
For virtual packages, we create a sanitized name from the path:
- owner/repo/prompts/code-review.prompt.md -> repo-code-review
- owner/repo/collections/project-planning -> repo-project-planning
- owner/repo/collections/project-planning.collection.yml -> repo-project-planning
"""
if not self.is_virtual or not self.virtual_path:
return self.repo_url.split("/")[-1] # Return repo name as fallback
# Extract repo name and file/collection name
repo_parts = self.repo_url.split("/")
repo_name = repo_parts[-1] if repo_parts else "package"
# Get the basename without extension
path_parts = self.virtual_path.split("/")
if self.is_virtual_collection():
# For collections: use the collection name without extension
# collections/project-planning -> project-planning
# collections/project-planning.collection.yml -> project-planning
collection_name = path_parts[-1]
# Strip .collection.yml/.collection.yaml extension if present
for ext in (".collection.yml", ".collection.yaml"):
if collection_name.endswith(ext):
collection_name = collection_name[: -len(ext)]
break
return f"{repo_name}-{collection_name}"
else:
# For individual files: use the filename without extension
# prompts/code-review.prompt.md -> code-review
filename = path_parts[-1]
for ext in self.VIRTUAL_FILE_EXTENSIONS:
if filename.endswith(ext):
filename = filename[: -len(ext)]
break
return f"{repo_name}-{filename}"
@staticmethod
def is_local_path(dep_str: str) -> bool:
"""Check if a dependency string looks like a local filesystem path.
Local paths start with './', '../', '/', '~/', '~\\', or a Windows drive
letter (e.g. 'C:\\' or 'C:/').
Protocol-relative URLs ('//...') are explicitly excluded.
"""
s = dep_str.strip()
# Reject protocol-relative URLs ('//...')
if s.startswith("//"):
return False
if s.startswith(('./','../', '/', '~/', '~\\', '.\\', '..\\')):
return True
# Windows absolute paths: drive letter + colon + separator (C:\ or C:/).
# Only ASCII letters A-Z/a-z are valid drive letters.
if (
len(s) >= 3
and (('A' <= s[0] <= 'Z') or ('a' <= s[0] <= 'z'))
and s[1] == ':'
and s[2] in ('\\', '/')
):
return True
return False
def get_unique_key(self) -> str:
"""Get a unique key for this dependency for deduplication.
For regular packages: repo_url
For virtual packages: repo_url + virtual_path to ensure uniqueness
For local packages: the local_path
Returns:
str: Unique key for this dependency
"""
if self.is_local and self.local_path:
return self.local_path
if self.is_virtual and self.virtual_path:
return f"{self.repo_url}/{self.virtual_path}"
return self.repo_url
def to_canonical(self) -> str:
"""Return the canonical form of this dependency for storage in apm.yml.
Follows the Docker-style default-registry convention:
- Default host (github.com) is stripped -> owner/repo
- Non-default hosts are preserved -> gitlab.com/owner/repo
- Virtual paths are appended -> owner/repo/path/to/thing
- Refs are appended with # -> owner/repo#v1.0
- Local paths are returned as-is -> ./packages/my-pkg
No .git suffix, no https://, no git@ -- just the canonical identifier.
Returns:
str: Canonical dependency string
"""
if self.is_local and self.local_path:
return self.local_path
host = self.host or default_host()
is_default = host.lower() == default_host().lower()
# Start with optional host prefix
if is_default and not self.artifactory_prefix:
result = self.repo_url
elif self.artifactory_prefix:
result = f"{host}/{self.artifactory_prefix}/{self.repo_url}"
else:
result = f"{host}/{self.repo_url}"
# Append virtual path for virtual packages
if self.is_virtual and self.virtual_path:
result = f"{result}/{self.virtual_path}"
# Append reference (branch, tag, commit)
if self.reference:
result = f"{result}#{self.reference}"
return result
def get_identity(self) -> str:
"""Return the identity of this dependency (canonical form without ref/alias).
Two deps with the same identity are the same package, regardless of
which ref or alias they specify. Used for duplicate detection and uninstall matching.
Returns:
str: Identity string (e.g., "owner/repo" or "gitlab.com/owner/repo/path")
"""
if self.is_local and self.local_path:
return self.local_path
host = self.host or default_host()
is_default = host.lower() == default_host().lower()
if is_default and not self.artifactory_prefix:
result = self.repo_url
elif self.artifactory_prefix:
result = f"{host}/{self.artifactory_prefix}/{self.repo_url}"
else:
result = f"{host}/{self.repo_url}"
if self.is_virtual and self.virtual_path:
result = f"{result}/{self.virtual_path}"
return result
@staticmethod
def canonicalize(raw: str) -> str:
"""Parse any raw input form and return its canonical storage form.
Convenience method that combines parse() + to_canonical().
Args:
raw: Any supported input form (shorthand, FQDN, HTTPS, SSH, etc.)
Returns:
str: Canonical form for apm.yml storage
"""
return DependencyReference.parse(raw).to_canonical()
def get_canonical_dependency_string(self) -> str:
"""Get the host-blind canonical string for filesystem and orphan-detection matching.
This returns repo_url (+ virtual_path) without host prefix -- it matches
the filesystem layout in apm_modules/ which is also host-blind.
For identity-based matching that includes non-default hosts, use get_identity().
For the full canonical form suitable for apm.yml storage, use to_canonical().
Returns:
str: Host-blind canonical string (e.g., "owner/repo")
"""
return self.get_unique_key()
def get_install_path(self, apm_modules_dir: Path) -> Path:
"""Get the canonical filesystem path where this package should be installed.
This is the single source of truth for where a package lives in apm_modules/.
For regular packages:
- GitHub: apm_modules/owner/repo/
- ADO: apm_modules/org/project/repo/
For virtual file/collection packages:
- GitHub: apm_modules/owner/<virtual-package-name>/
- ADO: apm_modules/org/project/<virtual-package-name>/
For subdirectory packages (Claude Skills, nested APM packages):
- GitHub: apm_modules/owner/repo/subdir/path/
- ADO: apm_modules/org/project/repo/subdir/path/
For local packages:
- apm_modules/_local/<directory-name>/
Args:
apm_modules_dir: Path to the apm_modules directory
Raises:
PathTraversalError: If the computed path escapes apm_modules_dir
Returns:
Path: Absolute path to the package installation directory
"""
if self.is_local and self.local_path:
pkg_dir_name = Path(self.local_path).name
validate_path_segments(
pkg_dir_name,
context="local package path",
reject_empty=True,
)
result = apm_modules_dir / "_local" / pkg_dir_name
ensure_path_within(result, apm_modules_dir)
return result
repo_parts = self.repo_url.split("/")
# Security: reject traversal in repo_url segments (catches lockfile injection)
validate_path_segments(self.repo_url, context="repo_url")
# Security: reject traversal in virtual_path (catches lockfile injection)
if self.virtual_path:
validate_path_segments(self.virtual_path, context="virtual_path")
result: Path | None = None
if self.is_virtual:
# Subdirectory packages (like Claude Skills) should use natural path structure
if self.is_virtual_subdirectory():
# Use repo path + subdirectory path
if self.is_azure_devops() and len(repo_parts) >= 3:
# ADO: org/project/repo/subdir
result = (
apm_modules_dir
/ repo_parts[0]
/ repo_parts[1]
/ repo_parts[2]
/ self.virtual_path
)
elif len(repo_parts) >= 2:
# owner/repo/subdir or group/subgroup/repo/subdir
result = apm_modules_dir.joinpath(*repo_parts, self.virtual_path)
else:
# Virtual file/collection: use sanitized package name (flattened)
package_name = self.get_virtual_package_name()
if self.is_azure_devops() and len(repo_parts) >= 3:
# ADO: org/project/virtual-pkg-name
result = (
apm_modules_dir / repo_parts[0] / repo_parts[1] / package_name
)
elif len(repo_parts) >= 2:
# owner/virtual-pkg-name (use first segment as namespace)
result = apm_modules_dir / repo_parts[0] / package_name
else:
# Regular package: use full repo path
if self.is_azure_devops() and len(repo_parts) >= 3:
# ADO: org/project/repo
result = apm_modules_dir / repo_parts[0] / repo_parts[1] / repo_parts[2]
elif len(repo_parts) >= 2:
# owner/repo or group/subgroup/repo (generic hosts)
result = apm_modules_dir.joinpath(*repo_parts)
if result is None:
# Fallback: join all parts
result = apm_modules_dir.joinpath(*repo_parts)
# Security: ensure the computed path stays within apm_modules/
ensure_path_within(result, apm_modules_dir)
return result
@staticmethod
def _normalize_ssh_protocol_url(url: str) -> str:
"""Normalize ssh:// protocol URLs to git@ format for consistent parsing.
Converts:
- ssh://git@gitlab.com/owner/repo.git -> git@gitlab.com:owner/repo.git
- ssh://git@host:port/owner/repo.git -> git@host:owner/repo.git
Non-SSH URLs are returned unchanged.
"""
if not url.startswith("ssh://"):
return url
# Parse the ssh:// URL
# Format: ssh://[user@]host[:port]/path
remainder = url[6:] # Remove 'ssh://'
# Extract user if present (typically 'git@')
user_prefix = ""
if "@" in remainder.split("/")[0]:
user_at_idx = remainder.index("@")
user_prefix = remainder[: user_at_idx + 1] # e.g., "git@"
remainder = remainder[user_at_idx + 1 :]
# Extract host (and optional port)
slash_idx = remainder.find("/")
if slash_idx == -1:
return url # Invalid format, return as-is
host_part = remainder[:slash_idx]
path_part = remainder[slash_idx + 1 :]
# Strip port if present (e.g., host:22)
if ":" in host_part:
host_part = host_part.split(":")[0]
# Convert to git@ format: git@host:path
if user_prefix:
return f"{user_prefix}{host_part}:{path_part}"
else:
return f"git@{host_part}:{path_part}"
@classmethod
def parse_from_dict(cls, entry: dict) -> "DependencyReference":
"""Parse an object-style dependency entry from apm.yml.
Supports the Cargo-inspired object format:
- git: https://gitlab.com/acme/coding-standards.git
path: instructions/security
ref: v2.0
- git: git@bitbucket.org:team/rules.git
path: prompts/review.prompt.md
Also supports local path entries:
- path: ./packages/my-shared-skills
Args:
entry: Dictionary with 'git' or 'path' (required), plus optional fields
Returns:
DependencyReference: Parsed dependency reference
Raises:
ValueError: If the entry is missing required fields or has invalid format
"""
# Support dict-form local path: { path: ./local/dir }
if "path" in entry and "git" not in entry:
local = entry["path"]
if not isinstance(local, str) or not local.strip():
raise ValueError("'path' field must be a non-empty string")
local = local.strip()
if not cls.is_local_path(local):
raise ValueError(
f"Object-style dependency must have a 'git' field, "
f"or 'path' must be a local filesystem path "
f"(starting with './', '../', '/', or '~')"
)
return cls.parse(local)
if "git" not in entry:
raise ValueError(
"Object-style dependency must have a 'git' or 'path' field"
)
git_url = entry["git"]
if not isinstance(git_url, str) or not git_url.strip():
raise ValueError("'git' field must be a non-empty string")
sub_path = entry.get("path")
ref_override = entry.get("ref")
alias_override = entry.get("alias")
# Validate sub_path if provided
if sub_path is not None:
if not isinstance(sub_path, str) or not sub_path.strip():
raise ValueError("'path' field must be a non-empty string")
sub_path = sub_path.strip().strip("/")
# Normalize backslashes to forward slashes for cross-platform safety
sub_path = sub_path.replace("\\", "/").strip().strip("/")
# Security: reject path traversal
validate_path_segments(sub_path, context="path")
# Parse the git URL using the standard parser
dep = cls.parse(git_url)
# Apply overrides from the object fields
if ref_override is not None:
if not isinstance(ref_override, str) or not ref_override.strip():
raise ValueError("'ref' field must be a non-empty string")
dep.reference = ref_override.strip()
if alias_override is not None:
if not isinstance(alias_override, str) or not alias_override.strip():
raise ValueError("'alias' field must be a non-empty string")
alias_override = alias_override.strip()
if not re.match(r"^[a-zA-Z0-9._-]+$", alias_override):
raise ValueError(
f"Invalid alias: {alias_override}. Aliases can only contain letters, numbers, dots, underscores, and hyphens"
)
dep.alias = alias_override
# Apply sub-path as virtual package
if sub_path:
dep.virtual_path = sub_path
dep.is_virtual = True
return dep
@classmethod
def _detect_virtual_package(cls, dependency_str: str):
"""Detect whether *dependency_str* refers to a virtual package.
Returns:
(is_virtual_package, virtual_path, validated_host)
"""
# Temporarily remove reference for path segment counting
temp_str = dependency_str
if "#" in temp_str:
temp_str = temp_str.rsplit("#", 1)[0]
is_virtual_package = False
virtual_path = None
validated_host = None
if temp_str.startswith(("git@", "https://", "http://")):
return is_virtual_package, virtual_path, validated_host
check_str = temp_str
if "/" in check_str:
first_segment = check_str.split("/")[0]
if "." in first_segment:
test_url = f"https://{check_str}"
try:
parsed = urllib.parse.urlparse(test_url)
hostname = parsed.hostname
if hostname and is_supported_git_host(hostname):
validated_host = hostname
path_parts = parsed.path.lstrip("/").split("/")
if len(path_parts) >= 2:
check_str = "/".join(check_str.split("/")[1:])
else:
raise ValueError(
unsupported_host_error(hostname or first_segment)
)
except (ValueError, AttributeError) as e:
if isinstance(e, ValueError) and "Invalid Git host" in str(e):
raise
raise ValueError(unsupported_host_error(first_segment))
elif check_str.startswith("gh/"):
check_str = "/".join(check_str.split("/")[1:])
path_segments = [seg for seg in check_str.split("/") if seg]
is_ado = validated_host is not None and is_azure_devops_hostname(validated_host)
is_generic_host = (
validated_host is not None
and not is_github_hostname(validated_host)
and not is_azure_devops_hostname(validated_host)
)
if is_ado and "_git" in path_segments:
git_idx = path_segments.index("_git")
path_segments = path_segments[:git_idx] + path_segments[git_idx + 1 :]
# Detect Artifactory VCS paths (artifactory/{repo-key}/{owner}/{repo})
is_artifactory = is_generic_host and is_artifactory_path(path_segments)
if is_ado:
min_base_segments = 3
elif is_artifactory:
# Artifactory: artifactory/{repo-key}/{owner}/{repo}
min_base_segments = 4
elif is_generic_host:
has_virtual_ext = any(
any(seg.endswith(ext) for ext in cls.VIRTUAL_FILE_EXTENSIONS)
for seg in path_segments
)
has_collection = "collections" in path_segments
if has_virtual_ext or has_collection:
min_base_segments = 2
else:
min_base_segments = len(path_segments)
else:
min_base_segments = 2
min_virtual_segments = min_base_segments + 1
if len(path_segments) >= min_virtual_segments:
is_virtual_package = True
virtual_path = "/".join(path_segments[min_base_segments:])
# Security: reject path traversal in virtual path
validate_path_segments(virtual_path, context="virtual path")
if "/collections/" in check_str or virtual_path.startswith("collections/"):
pass
elif any(virtual_path.endswith(ext) for ext in cls.VIRTUAL_FILE_EXTENSIONS):
pass
else:
last_segment = virtual_path.split("/")[-1]
if "." in last_segment:
raise InvalidVirtualPackageExtensionError(
f"Invalid virtual package path '{virtual_path}'. "
f"Individual files must end with one of: {', '.join(cls.VIRTUAL_FILE_EXTENSIONS)}. "
f"For subdirectory packages, the path should not have a file extension."
)
return is_virtual_package, virtual_path, validated_host
@staticmethod
def _parse_ssh_url(dependency_str: str):
"""Parse an SSH-style git URL (``git@host:owner/repo``).
Returns:
``(host, repo_url, reference, alias)`` or *None* if not an SSH URL.
"""
ssh_match = re.match(r"^git@([^:]+):(.+)$", dependency_str)
if not ssh_match:
return None
host = ssh_match.group(1)
ssh_repo_part = ssh_match.group(2)
reference = None
alias = None
if "@" in ssh_repo_part:
ssh_repo_part, alias = ssh_repo_part.rsplit("@", 1)
alias = alias.strip()
if "#" in ssh_repo_part:
repo_part, reference = ssh_repo_part.rsplit("#", 1)
reference = reference.strip()
else:
repo_part = ssh_repo_part
if repo_part.endswith(".git"):
repo_part = repo_part[:-4]
repo_url = repo_part.strip()
# Security: reject traversal sequences in SSH repo paths
validate_path_segments(
repo_url, context="SSH repository path", reject_empty=True
)
return host, repo_url, reference, alias
@classmethod
def _parse_standard_url(
cls, dependency_str: str, is_virtual_package: bool, virtual_path, validated_host
):
"""Parse a non-SSH dependency string (HTTPS, FQDN, or shorthand).
Returns:
``(host, repo_url, reference, alias)``
"""
host = None
alias = None
reference = None
if "#" in dependency_str:
repo_part, reference = dependency_str.rsplit("#", 1)
reference = reference.strip()
else:
repo_part = dependency_str
repo_url = repo_part.strip()
# For virtual packages, extract just the owner/repo part (or org/project/repo for ADO)
if is_virtual_package and not repo_url.startswith(("https://", "http://")):
parts = repo_url.split("/")
if "_git" in parts:
git_idx = parts.index("_git")
parts = parts[:git_idx] + parts[git_idx + 1 :]
if len(parts) >= 3 and is_supported_git_host(parts[0]):
host = parts[0]
if is_azure_devops_hostname(parts[0]):
if len(parts) < 5:
raise ValueError(
"Invalid Azure DevOps virtual package format: must be dev.azure.com/org/project/repo/path"
)
repo_url = "/".join(parts[1:4])
elif is_artifactory_path(parts[1:]):
art_result = parse_artifactory_path(parts[1:])
if art_result:
repo_url = f"{art_result[1]}/{art_result[2]}"
else:
repo_url = "/".join(parts[1:3])
elif len(parts) >= 2:
if not host:
host = default_host()
if validated_host and is_azure_devops_hostname(validated_host):
if len(parts) < 4:
raise ValueError(
"Invalid Azure DevOps virtual package format: expected at least org/project/repo/path"
)
repo_url = "/".join(parts[:3])
else:
repo_url = "/".join(parts[:2])
# Normalize to URL format for secure parsing
if repo_url.startswith(("https://", "http://")):
parsed_url = urllib.parse.urlparse(repo_url)
host = parsed_url.hostname or ""
else:
parts = repo_url.split("/")
if "_git" in parts:
git_idx = parts.index("_git")
parts = parts[:git_idx] + parts[git_idx + 1 :]
if len(parts) >= 3 and is_supported_git_host(parts[0]):
host = parts[0]
if is_azure_devops_hostname(host) and len(parts) >= 4:
user_repo = "/".join(parts[1:4])
elif not is_github_hostname(host) and not is_azure_devops_hostname(
host
):
if is_artifactory_path(parts[1:]):
art_result = parse_artifactory_path(parts[1:])
if art_result:
user_repo = f"{art_result[1]}/{art_result[2]}"
else:
user_repo = "/".join(parts[1:])
else:
user_repo = "/".join(parts[1:])
else:
user_repo = "/".join(parts[1:3])
elif len(parts) >= 2 and "." not in parts[0]:
if not host:
host = default_host()
if is_azure_devops_hostname(host) and len(parts) >= 3:
user_repo = "/".join(parts[:3])
elif (
host
and not is_github_hostname(host)
and not is_azure_devops_hostname(host)
):
user_repo = "/".join(parts)
else:
user_repo = "/".join(parts[:2])
else:
raise ValueError(
f"Use 'user/repo' or 'github.com/user/repo' or 'dev.azure.com/org/project/repo' format"
)
if not user_repo or "/" not in user_repo:
raise ValueError(
f"Invalid repository format: {repo_url}. Expected 'user/repo' or 'org/project/repo'"
)
uparts = user_repo.split("/")
is_ado_host = host and is_azure_devops_hostname(host)
if is_ado_host:
if len(uparts) < 3:
raise ValueError(
f"Invalid Azure DevOps repository format: {repo_url}. Expected 'org/project/repo'"
)
else:
if len(uparts) < 2:
raise ValueError(
f"Invalid repository format: {repo_url}. Expected 'user/repo'"
)
allowed_pattern = (
r"^[a-zA-Z0-9._\- ]+$" if is_ado_host else r"^[a-zA-Z0-9._-]+$"
)
validate_path_segments(
"/".join(uparts), context="repository path"
)
for part in uparts:
if not re.match(allowed_pattern, part.rstrip(".git")):
raise ValueError(f"Invalid repository path component: {part}")
quoted_repo = "/".join(urllib.parse.quote(p, safe="") for p in uparts)
github_url = urllib.parse.urljoin(f"https://{host}/", quoted_repo)
parsed_url = urllib.parse.urlparse(github_url)
hostname = parsed_url.hostname or ""
if not is_supported_git_host(hostname):
raise ValueError(unsupported_host_error(hostname or parsed_url.netloc))
path = parsed_url.path.strip("/")
if not path:
raise ValueError("Repository path cannot be empty")
if path.endswith(".git"):
path = path[:-4]
path_parts = [urllib.parse.unquote(p) for p in path.split("/")]
if "_git" in path_parts:
git_idx = path_parts.index("_git")
path_parts = path_parts[:git_idx] + path_parts[git_idx + 1 :]
is_ado_host = is_azure_devops_hostname(hostname)
if is_ado_host:
if len(path_parts) != 3:
raise ValueError(
f"Invalid Azure DevOps repository path: expected 'org/project/repo', got '{path}'"
)
else:
if len(path_parts) < 2:
raise ValueError(
f"Invalid repository path: expected at least 'user/repo', got '{path}'"
)
for pp in path_parts:
if any(pp.endswith(ext) for ext in cls.VIRTUAL_FILE_EXTENSIONS):
raise ValueError(
f"Invalid repository path: '{path}' contains a virtual file extension. "
f"Use the dict format with 'path:' for virtual packages in HTTPS URLs"
)
allowed_pattern = (
r"^[a-zA-Z0-9._\- ]+$" if is_ado_host else r"^[a-zA-Z0-9._-]+$"
)
validate_path_segments(
"/".join(path_parts),
context="repository URL path",
reject_empty=True,
)
for part in path_parts:
if not re.match(allowed_pattern, part):
raise ValueError(f"Invalid repository path component: {part}")
repo_url = "/".join(path_parts)
if not host:
host = default_host()
return host, repo_url, reference, alias
@classmethod
def parse(cls, dependency_str: str) -> "DependencyReference":
"""Parse a dependency string into a DependencyReference.
Supports formats:
- user/repo
- user/repo#branch
- user/repo#v1.0.0
- user/repo#commit_sha
- github.com/user/repo#ref
- user/repo@alias
- user/repo#ref@alias
- user/repo/path/to/file.prompt.md (virtual file package)
- user/repo/collections/name (virtual collection package)
- https://gitlab.com/owner/repo.git (generic HTTPS git URL)
- git@gitlab.com:owner/repo.git (SSH git URL)
- ssh://git@gitlab.com/owner/repo.git (SSH protocol URL)
- ./local/path (local filesystem path)
- /absolute/path (local filesystem path)
- ../relative/path (local filesystem path)
Any valid FQDN is accepted as a git host (GitHub, GitLab, Bitbucket,
self-hosted instances, etc.).
Args:
dependency_str: The dependency string to parse
Returns:
DependencyReference: Parsed dependency reference
Raises:
ValueError: If the dependency string format is invalid
"""
if not dependency_str.strip():
raise ValueError("Empty dependency string")
dependency_str = urllib.parse.unquote(dependency_str)
if any(ord(c) < 32 for c in dependency_str):
raise ValueError("Dependency string contains invalid control characters")
# --- Local path detection (must run before URL/host parsing) ---
if cls.is_local_path(dependency_str):
local = dependency_str.strip()
pkg_name = Path(local).name
if not pkg_name or pkg_name in (".", ".."):
raise ValueError(
f"Local path '{local}' does not resolve to a named directory. "
f"Use a path that ends with a directory name "
f"(e.g., './my-package' instead of './')."
)
return cls(
repo_url=f"_local/{pkg_name}",
is_local=True,
local_path=local,
)
if dependency_str.startswith("//"):
raise ValueError(
unsupported_host_error(
"//...", context="Protocol-relative URLs are not supported"
)
)
# Preserve the original ssh:// URL verbatim before normalization so the
# downloader can clone with the exact user-supplied URL (e.g. custom port).
original_ssh_url = dependency_str if dependency_str.startswith("ssh://") else None
dependency_str = cls._normalize_ssh_protocol_url(dependency_str)
# Phase 1: detect virtual packages
is_virtual_package, virtual_path, validated_host = cls._detect_virtual_package(
dependency_str
)
# Phase 2: parse SSH or standard URL
ssh_result = cls._parse_ssh_url(dependency_str)
if ssh_result:
host, repo_url, reference, alias = ssh_result
else:
host, repo_url, reference, alias = cls._parse_standard_url(
dependency_str, is_virtual_package, virtual_path, validated_host
)
# Phase 3: final validation and ADO field extraction
is_ado_final = host and is_azure_devops_hostname(host)
if is_ado_final:
if not re.match(
r"^[a-zA-Z0-9._-]+/[a-zA-Z0-9._\- ]+/[a-zA-Z0-9._\- ]+$", repo_url
):
raise ValueError(
f"Invalid Azure DevOps repository format: {repo_url}. Expected 'org/project/repo'"
)
ado_parts = repo_url.split("/")
validate_path_segments(
repo_url, context="Azure DevOps repository path"
)
ado_organization = ado_parts[0]
ado_project = ado_parts[1]
ado_repo = ado_parts[2]
else:
segments = repo_url.split("/")
if len(segments) < 2:
raise ValueError(
f"Invalid repository format: {repo_url}. Expected 'user/repo'"
)
if not all(re.match(r"^[a-zA-Z0-9._-]+$", s) for s in segments):
raise ValueError(
f"Invalid repository format: {repo_url}. Contains invalid characters"
)
validate_path_segments(repo_url, context="repository path")
for seg in segments:
if any(seg.endswith(ext) for ext in cls.VIRTUAL_FILE_EXTENSIONS):
raise ValueError(
f"Invalid repository format: '{repo_url}' contains a virtual file extension. "
f"Use the dict format with 'path:' for virtual packages in SSH/HTTPS URLs"
)
ado_organization = None
ado_project = None
ado_repo = None
if alias and not re.match(r"^[a-zA-Z0-9._-]+$", alias):
raise ValueError(
f"Invalid alias: {alias}. Aliases can only contain letters, numbers, dots, underscores, and hyphens"
)
# Extract Artifactory prefix from the original path if applicable
artifactory_prefix = None
if host and not is_ado_final:
_art_str = dependency_str.split("#")[0].split("@")[0]
# Strip scheme if present (e.g., https://host/artifactory/...)
if "://" in _art_str:
_art_str = _art_str.split("://", 1)[1]
_art_segs = _art_str.replace(f"{host}/", "", 1).split("/")
if is_artifactory_path(_art_segs):
art_result = parse_artifactory_path(_art_segs)
if art_result:
artifactory_prefix = art_result[0]
return cls(
repo_url=repo_url,
host=host,
reference=reference,
alias=alias,
virtual_path=virtual_path,
is_virtual=is_virtual_package,
ado_organization=ado_organization,
ado_project=ado_project,
ado_repo=ado_repo,
artifactory_prefix=artifactory_prefix,
original_ssh_url=original_ssh_url,
)