-
Notifications
You must be signed in to change notification settings - Fork 83
Expand file tree
/
Copy pathupload.py
More file actions
143 lines (121 loc) · 5.56 KB
/
upload.py
File metadata and controls
143 lines (121 loc) · 5.56 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
import time
from datetime import datetime, timezone
from django.db import transaction
from django.contrib.sessions.models import Session
from pydantic import TypeAdapter
from pulpcore.plugin.models import Artifact, CreatedResource, Content, ContentArtifact
from pulpcore.plugin.util import get_domain, get_current_authenticated_user, get_prn
from pulp_python.app.models import PythonPackageContent, PythonRepository, PackageProvenance
from pulp_python.app.provenance import (
Attestation,
AttestationBundle,
AnyPublisher,
Provenance,
verify_provenance,
)
from pulp_python.app.utils import artifact_to_metadata_artifact, artifact_to_python_content_data
def upload(artifact_sha256, filename, attestations=None, repository_pk=None):
"""
Uploads a Python Package to Pulp
Args:
artifact_sha256: the sha256 of the artifact in Pulp to create a package from
filename: the full filename of the package to create
attestations: optional list of attestations to create a provenance from
repository_pk: the optional pk of the repository to add the content to
"""
domain = get_domain()
pre_check = PythonPackageContent.objects.filter(sha256=artifact_sha256, _pulp_domain=domain)
content_to_add = [pre_check.first() or create_content(artifact_sha256, filename, domain)]
if attestations:
content_to_add += [create_provenance(content_to_add[0], attestations, domain)]
content_to_add = Content.objects.filter(pk__in=[c.pk for c in content_to_add])
content_to_add.touch()
if repository_pk:
repository = PythonRepository.objects.get(pk=repository_pk)
with repository.new_version() as new_version:
new_version.add_content(content_to_add)
def upload_group(session_pk, repository_pk=None):
"""
Uploads a Python Package to Pulp
Args:
session_pk: the session that has the artifacts to upload
repository_pk: optional repository to add Content to
"""
s_query = Session.objects.select_for_update().filter(pk=session_pk)
domain = get_domain()
while True:
with transaction.atomic():
session_data = s_query.first().get_decoded()
now = datetime.now(tz=timezone.utc)
start_time = datetime.fromisoformat(session_data["start"])
if now >= start_time:
content_to_add = Content.objects.none()
for artifact_sha256, filename, attestations in session_data["artifacts"]:
pre_check = PythonPackageContent.objects.filter(
sha256=artifact_sha256, _pulp_domain=domain
).first()
content = [pre_check or create_content(artifact_sha256, filename, domain)]
if attestations:
content += [create_provenance(content[0], attestations, domain)]
content = Content.objects.filter(pk__in=[c.pk for c in content])
content.touch()
content_to_add |= content
if repository_pk:
repository = PythonRepository.objects.get(pk=repository_pk)
with repository.new_version() as new_version:
new_version.add_content(content_to_add)
return
else:
sleep_time = start_time - now
time.sleep(sleep_time.seconds)
def create_content(artifact_sha256, filename, domain):
"""
Creates PythonPackageContent from artifact.
Args:
artifact_sha256: validated artifact
filename: file name
domain: the pulp_domain to perform this task in
Returns:
the newly created PythonPackageContent
"""
artifact = Artifact.objects.get(sha256=artifact_sha256, pulp_domain=domain)
data = artifact_to_python_content_data(filename, artifact, domain)
@transaction.atomic()
def create():
content = PythonPackageContent.objects.create(**data)
ContentArtifact.objects.create(artifact=artifact, content=content, relative_path=filename)
if metadata_artifact := artifact_to_metadata_artifact(filename, artifact):
ContentArtifact.objects.create(
artifact=metadata_artifact, content=content, relative_path=f"{filename}.metadata"
)
return content
new_content = create()
resource = CreatedResource(content_object=new_content)
resource.save()
return new_content
def create_provenance(package, attestations, domain):
"""
Creates PackageProvenance from attestations.
Args:
package: the package to create the provenance for
attestations: the attestations to create the provenance from
domain: the pulp_domain to perform this task in
Returns:
the newly created PackageProvenance
"""
attestations = TypeAdapter(list[Attestation]).validate_python(attestations)
user = get_current_authenticated_user()
publisher = AnyPublisher(kind="Pulp User", prn=get_prn(user))
att_bundle = AttestationBundle(publisher=publisher, attestations=attestations)
provenance = Provenance(attestation_bundles=[att_bundle])
verify_provenance(package.filename, package.sha256, provenance)
provenance_json = provenance.model_dump(mode="json")
prov_sha256 = PackageProvenance.calculate_sha256(provenance_json)
prov_model, _ = PackageProvenance.objects.get_or_create(
sha256=prov_sha256,
_pulp_domain=domain,
defaults={"package": package, "provenance": provenance_json},
)
resource = CreatedResource(content_object=prov_model)
resource.save()
return prov_model