|
1
|
+ |
defmodule OpenAgents.Deployments do
|
|
2
|
+ |
@moduledoc """
|
|
3
|
+ |
The deployment control plane: one place that decides whether exact bytes may
|
|
4
|
+ |
reach a repository's environment, and records why.
|
|
5
|
+ |
|
|
6
|
+ |
## What this context owns
|
|
7
|
+ |
|
|
8
|
+ |
* Repository-scoped environments and their protection policy.
|
|
9
|
+ |
* Durable deployment requests: an intent for one exact commit and artifact.
|
|
10
|
+ |
* Durable runs: the admitted execution of one request, with a lifecycle no
|
|
11
|
+ |
caller can shortcut.
|
|
12
|
+ |
* Approvals, check results, and an append-only event stream.
|
|
13
|
+ |
|
|
14
|
+ |
## The rules that hold everywhere
|
|
15
|
+ |
|
|
16
|
+ |
* Authority comes from the principal, never from the request body. See
|
|
17
|
+ |
`OpenAgents.Deployments.Authority`.
|
|
18
|
+ |
* Every transition is checked against `OpenAgents.Deployments.Lifecycle` and
|
|
19
|
+ |
written transactionally with its event, so a run's state and its history
|
|
20
|
+ |
cannot disagree.
|
|
21
|
+ |
* Policy is re-evaluated, and authority re-checked, at sensitive transitions.
|
|
22
|
+ |
Membership is revoked and environments freeze between `requested` and
|
|
23
|
+ |
`deploying`.
|
|
24
|
+ |
* A request is idempotent on `{repository, environment, idempotency_key}`.
|
|
25
|
+ |
Replaying a key with the same bytes returns the original run; replaying it
|
|
26
|
+ |
with different bytes is a conflict, not a second deployment.
|
|
27
|
+ |
* Durable records hold secret *references*. Values are resolved at execution
|
|
28
|
+ |
time, for the bound provider and environment, and never stored.
|
|
29
|
+ |
* Platform-operator authority recovers the control plane. It does not deploy
|
|
30
|
+ |
tenant code, and `deployments:promote` — the Forge fleet promotion scope —
|
|
31
|
+ |
is never exposed to a repository workflow.
|
|
32
|
+ |
"""
|
|
33
|
+ |
|
|
34
|
+ |
import Ecto.Query
|
|
35
|
+ |
|
|
36
|
+ |
alias Ecto.Multi
|
|
37
|
+ |
alias OpenAgents.Accounts.User
|
|
38
|
+ |
alias OpenAgents.Deployments.Approval
|
|
39
|
+ |
alias OpenAgents.Deployments.Authority
|
|
40
|
+ |
alias OpenAgents.Deployments.CheckResult
|
|
41
|
+ |
alias OpenAgents.Deployments.Environment
|
|
42
|
+ |
alias OpenAgents.Deployments.Event
|
|
43
|
+ |
alias OpenAgents.Deployments.Lifecycle
|
|
44
|
+ |
alias OpenAgents.Deployments.Policy
|
|
45
|
+ |
alias OpenAgents.Deployments.Principal
|
|
46
|
+ |
alias OpenAgents.Deployments.Protection
|
|
47
|
+ |
alias OpenAgents.Deployments.Provider
|
|
48
|
+ |
alias OpenAgents.Deployments.Request
|
|
49
|
+ |
alias OpenAgents.Deployments.Run
|
|
50
|
+ |
alias OpenAgents.Deployments.WorkflowGrant
|
|
51
|
+ |
alias OpenAgents.Repo
|
|
52
|
+ |
alias OpenAgents.Repositories.Repository
|
|
53
|
+ |
|
|
54
|
+ |
@default_limit 25
|
|
55
|
+ |
@maximum_limit 100
|
|
56
|
+ |
@default_lease_seconds 300
|
|
57
|
+ |
|
|
58
|
+ |
@type error ::
|
|
59
|
+ |
{:forbidden, Authority.reason()}
|
|
60
|
+ |
| :environment_not_found
|
|
61
|
+ |
| :run_not_found
|
|
62
|
+ |
| :request_not_found
|
|
63
|
+ |
| :unknown_provider
|
|
64
|
+ |
| :unknown_commit
|
|
65
|
+ |
| :idempotency_conflict
|
|
66
|
+ |
| :precondition_failed
|
|
67
|
+ |
| {:illegal_transition, String.t(), String.t()}
|
|
68
|
+ |
| {:policy_denied, String.t()}
|
|
69
|
+ |
| Ecto.Changeset.t()
|
|
70
|
+ |
|
|
71
|
+ |
# ---------------------------------------------------------------------------
|
|
72
|
+ |
# Environments
|
|
73
|
+ |
# ---------------------------------------------------------------------------
|
|
74
|
+ |
|
|
75
|
+ |
@doc """
|
|
76
|
+ |
Define or replace one environment's provider binding and protection policy.
|
|
77
|
+ |
|
|
78
|
+ |
Only a repository member with write authority can define an environment, and
|
|
79
|
+ |
the provider must be one the platform configures: a tenant that could name an
|
|
80
|
+ |
arbitrary module would choose which code the control plane runs.
|
|
81
|
+ |
"""
|
|
82
|
+ |
@spec put_environment(Repository.t(), Principal.t(), map()) ::
|
|
83
|
+ |
{:ok, Environment.t()} | {:error, error()}
|
|
84
|
+ |
def put_environment(%Repository{} = repository, %Principal{} = principal, attrs) do
|
|
85
|
+ |
attrs = normalize(attrs)
|
|
86
|
+ |
|
|
87
|
+ |
with :ok <- authorize_write(principal, repository),
|
|
88
|
+ |
{:ok, _module} <- Provider.fetch(attrs["provider"] || "") do
|
|
89
|
+ |
name = attrs["name"]
|
|
90
|
+ |
|
|
91
|
+ |
environment =
|
|
92
|
+ |
case fetch_environment_record(repository, name) do
|
|
93
|
+ |
{:ok, existing} -> existing
|
|
94
|
+ |
{:error, :environment_not_found} -> %Environment{repository_id: repository.id}
|
|
95
|
+ |
end
|
|
96
|
+ |
|
|
97
|
+ |
environment
|
|
98
|
+ |
|> Environment.changeset(Map.put_new(attrs, "protection", %{}))
|
|
99
|
+ |
|> put_creator(principal, environment)
|
|
100
|
+ |
|> Repo.insert_or_update()
|
|
101
|
+ |
end
|
|
102
|
+ |
end
|
|
103
|
+ |
|
|
104
|
+ |
@doc "List one repository's environments, with their protection requirements."
|
|
105
|
+ |
@spec list_environments(Repository.t(), Principal.t()) ::
|
|
106
|
+ |
{:ok, [Environment.t()]} | {:error, error()}
|
|
107
|
+ |
def list_environments(%Repository{} = repository, %Principal{} = principal) do
|
|
108
|
+ |
with :ok <- Authority.authorize_read(principal, repository) do
|
|
109
|
+ |
{:ok,
|
|
110
|
+ |
Repo.all(
|
|
111
|
+ |
from environment in Environment,
|
|
112
|
+ |
where: environment.repository_id == ^repository.id,
|
|
113
|
+ |
order_by: [asc: environment.name]
|
|
114
|
+ |
)}
|
|
115
|
+ |
end
|
|
116
|
+ |
end
|
|
117
|
+ |
|
|
118
|
+ |
@doc "Fetch one environment by name, enforcing repository visibility."
|
|
119
|
+ |
@spec fetch_environment(Repository.t(), Principal.t(), String.t()) ::
|
|
120
|
+ |
{:ok, Environment.t()} | {:error, error()}
|
|
121
|
+ |
def fetch_environment(%Repository{} = repository, %Principal{} = principal, name) do
|
|
122
|
+ |
with :ok <- Authority.authorize_read(principal, repository) do
|
|
123
|
+ |
fetch_environment_record(repository, name)
|
|
124
|
+ |
end
|
|
125
|
+ |
end
|
|
126
|
+ |
|
|
127
|
+ |
# ---------------------------------------------------------------------------
|
|
128
|
+ |
# Requests and runs
|
|
129
|
+ |
# ---------------------------------------------------------------------------
|
|
130
|
+ |
|
|
131
|
+ |
@doc """
|
|
132
|
+ |
Record a deployment intent and admit it as far as policy allows.
|
|
133
|
+ |
|
|
134
|
+ |
The request names exact bytes: a full commit sha and an artifact digest. The
|
|
135
|
+ |
commit is verified against the repository's own git storage through
|
|
136
|
+ |
`:commit_store`, so a request cannot deploy a sha the repository never
|
|
137
|
+ |
received.
|
|
138
|
+ |
|
|
139
|
+ |
On success the run holds the furthest state policy admits: `queued` when
|
|
140
|
+ |
nothing is outstanding, `checking` while required checks are missing,
|
|
141
|
+ |
`waiting_for_approval` while approvals are. A policy denial is durable too: the
|
|
142
|
+ |
run reaches `failed` with an explanation, because a deployment that was refused
|
|
143
|
+ |
and left no record is indistinguishable from one that was never asked for.
|
|
144
|
+ |
"""
|
|
145
|
+ |
@spec request_deployment(Repository.t(), Principal.t(), map(), keyword()) ::
|
|
146
|
+ |
{:ok, Run.t()} | {:error, error()}
|
|
147
|
+ |
def request_deployment(
|
|
148
|
+ |
%Repository{} = repository,
|
|
149
|
+ |
%Principal{} = principal,
|
|
150
|
+ |
attrs,
|
|
151
|
+ |
options \\ []
|
|
152
|
+ |
) do
|
|
153
|
+ |
attrs = normalize(attrs)
|
|
154
|
+ |
commit_store = Keyword.get_lazy(options, :commit_store, &default_commit_store/0)
|
|
155
|
+ |
|
|
156
|
+ |
with {:ok, environment} <- fetch_environment_record(repository, attrs["environment"]),
|
|
157
|
+ |
{:ok, principal_type} <- Principal.request_principal_type(principal),
|
|
158
|
+ |
intent = intent(attrs),
|
|
159
|
+ |
:ok <- Authority.authorize_request(principal, repository, environment, intent),
|
|
160
|
+ |
{:ok, provider} <- Provider.fetch(environment.provider),
|
|
161
|
+ |
:ok <- verify_commit(repository, attrs["commit_sha"], commit_store),
|
|
162
|
+ |
{:ok, request} <-
|
|
163
|
+ |
insert_request(repository, environment, principal, principal_type, attrs) do
|
|
164
|
+ |
case Repo.preload(request, :run) do
|
|
165
|
+ |
%Request{run: %Run{} = run} -> {:ok, run}
|
|
166
|
+ |
%Request{} -> admit(repository, environment, request, principal, provider)
|
|
167
|
+ |
end
|
|
168
|
+ |
end
|
|
169
|
+ |
end
|
|
170
|
+ |
|
|
171
|
+ |
@doc """
|
|
172
|
+ |
Fetch one run by id, enforcing repository visibility and boundary.
|
|
173
|
+ |
|
|
174
|
+ |
A valid run id from another repository is a `cross_repository` denial rather
|
|
175
|
+ |
than a successful read.
|
|
176
|
+ |
"""
|
|
177
|
+ |
@spec fetch_run(Repository.t(), Principal.t(), String.t()) ::
|
|
178
|
+ |
{:ok, Run.t()} | {:error, error()}
|
|
179
|
+ |
def fetch_run(%Repository{} = repository, %Principal{} = principal, id) do
|
|
180
|
+ |
with :ok <- Authority.authorize_read(principal, repository),
|
|
181
|
+ |
{:ok, run} <- fetch_run_record(id),
|
|
182
|
+ |
:ok <- Authority.check_repository(repository, run) do
|
|
183
|
+ |
{:ok, Repo.preload(run, [:deployment_request, :environment])}
|
|
184
|
+ |
end
|
|
185
|
+ |
end
|
|
186
|
+ |
|
|
187
|
+ |
@doc """
|
|
188
|
+ |
List a repository's runs, newest first, with bounded keyset pagination.
|
|
189
|
+ |
|
|
190
|
+ |
`:limit` is clamped to #{@maximum_limit}, and `:cursor` is the id of the last
|
|
191
|
+ |
run a caller has already seen.
|
|
192
|
+ |
"""
|
|
193
|
+ |
@spec list_runs(Repository.t(), Principal.t(), keyword()) ::
|
|
194
|
+ |
{:ok, [Run.t()]} | {:error, error()}
|
|
195
|
+ |
def list_runs(%Repository{} = repository, %Principal{} = principal, options \\ []) do
|
|
196
|
+ |
with :ok <- Authority.authorize_read(principal, repository) do
|
|
197
|
+ |
query =
|
|
198
|
+ |
from run in Run,
|
|
199
|
+ |
where: run.repository_id == ^repository.id,
|
|
200
|
+ |
order_by: [desc: run.inserted_at, desc: run.id],
|
|
201
|
+ |
limit: ^limit(options),
|
|
202
|
+ |
preload: [:deployment_request, :environment]
|
|
203
|
+ |
|
|
204
|
+ |
query =
|
|
205
|
+ |
case cursor_run(repository, options[:cursor]) do
|
|
206
|
+ |
nil ->
|
|
207
|
+ |
query
|
|
208
|
+ |
|
|
209
|
+ |
%Run{} = cursor ->
|
|
210
|
+ |
from run in query,
|
|
211
|
+ |
where:
|
|
212
|
+ |
run.inserted_at < ^cursor.inserted_at or
|
|
213
|
+ |
(run.inserted_at == ^cursor.inserted_at and run.id < ^cursor.id)
|
|
214
|
+ |
end
|
|
215
|
+ |
|
|
216
|
+ |
query =
|
|
217
|
+ |
case options[:environment_id] do
|
|
218
|
+ |
nil -> query
|
|
219
|
+ |
environment_id -> from run in query, where: run.environment_id == ^environment_id
|
|
220
|
+ |
end
|
|
221
|
+ |
|
|
222
|
+ |
query =
|
|
223
|
+ |
case options[:state] do
|
|
224
|
+ |
nil -> query
|
|
225
|
+ |
state -> from run in query, where: run.state == ^state
|
|
226
|
+ |
end
|
|
227
|
+ |
|
|
228
|
+ |
{:ok, Repo.all(query)}
|
|
229
|
+ |
end
|
|
230
|
+ |
end
|
|
231
|
+ |
|
|
232
|
+ |
@doc """
|
|
233
|
+ |
Request cancellation of a run.
|
|
234
|
+ |
|
|
235
|
+ |
Cancellation is bounded, not immediate. A run that has not reached the provider
|
|
236
|
+ |
cancels at once. A run already `deploying` records the request, and the worker
|
|
237
|
+ |
that holds the lease drives it to a terminal state itself, so the control plane
|
|
238
|
+ |
never claims a deployment stopped because someone asked.
|
|
239
|
+ |
"""
|
|
240
|
+ |
@spec cancel_run(Repository.t(), Principal.t(), String.t(), keyword()) ::
|
|
241
|
+ |
{:ok, Run.t()} | {:error, error()}
|
|
242
|
+ |
def cancel_run(%Repository{} = repository, %Principal{} = principal, id, options \\ []) do
|
|
243
|
+ |
with {:ok, run} <- fetch_run_record(id),
|
|
244
|
+ |
:ok <- Authority.check_repository(repository, run),
|
|
245
|
+ |
{:ok, request} <- fetch_request_record(run.deployment_request_id),
|
|
246
|
+ |
:ok <- Authority.authorize_cancel(principal, repository, run, request),
|
|
247
|
+ |
:ok <- precondition(run, options) do
|
|
248
|
+ |
cancel = fn ->
|
|
249
|
+ |
run
|
|
250
|
+ |
|> Ecto.Changeset.change(%{
|
|
251
|
+ |
cancel_requested_at: DateTime.utc_now(),
|
|
252
|
+ |
cancel_requested_by_user_id: cancel_requester(principal)
|
|
253
|
+ |
})
|
|
254
|
+ |
|> Repo.update()
|
|
255
|
+ |
end
|
|
256
|
+ |
|
|
257
|
+ |
cond do
|
|
258
|
+ |
Run.terminal?(run) ->
|
|
259
|
+ |
{:error, {:illegal_transition, run.state, "cancelled"}}
|
|
260
|
+ |
|
|
261
|
+ |
run.state == "deploying" ->
|
|
262
|
+ |
with {:ok, run} <- cancel.() do
|
|
263
|
+ |
{:ok, _event} = append_event(run, "cancellation_requested", principal, %{})
|
|
264
|
+ |
{:ok, run}
|
|
265
|
+ |
end
|
|
266
|
+ |
|
|
267
|
+ |
true ->
|
|
268
|
+ |
with {:ok, run} <- cancel.() do
|
|
269
|
+ |
transition(run, "cancelled", principal, reason: "cancelled_by_request")
|
|
270
|
+ |
end
|
|
271
|
+ |
end
|
|
272
|
+ |
end
|
|
273
|
+ |
end
|
|
274
|
+ |
|
|
275
|
+ |
# ---------------------------------------------------------------------------
|
|
276
|
+ |
# Approvals
|
|
277
|
+ |
# ---------------------------------------------------------------------------
|
|
278
|
+ |
|
|
279
|
+ |
@doc """
|
|
280
|
+ |
Record an approval decision and re-evaluate the run.
|
|
281
|
+ |
|
|
282
|
+ |
The decision is bound to the request digest it was made against, so it cannot
|
|
283
|
+ |
carry onto different bytes. The approver must hold a role the environment
|
|
284
|
+ |
admits and, under separation of duties, must not be the requester.
|
|
285
|
+ |
"""
|
|
286
|
+ |
@spec decide_run(Repository.t(), Principal.t(), String.t(), String.t(), map()) ::
|
|
287
|
+ |
{:ok, Run.t()} | {:error, error()}
|
|
288
|
+ |
def decide_run(%Repository{} = repository, %Principal{} = principal, id, decision, attrs \\ %{})
|
|
289
|
+ |
when decision in ~w(approved rejected) do
|
|
290
|
+ |
attrs = normalize(attrs)
|
|
291
|
+ |
|
|
292
|
+ |
with {:ok, run} <- fetch_run_record(id),
|
|
293
|
+ |
:ok <- Authority.check_repository(repository, run),
|
|
294
|
+ |
{:ok, request} <- fetch_request_record(run.deployment_request_id),
|
|
295
|
+ |
{:ok, environment} <- fetch_environment_by_id(repository, run.environment_id),
|
|
296
|
+ |
:ok <- Authority.authorize_approval(principal, repository, environment, request),
|
|
297
|
+ |
:ok <- approvable(run) do
|
|
298
|
+ |
%Approval{
|
|
299
|
+ |
repository_id: repository.id,
|
|
300
|
+ |
deployment_run_id: run.id,
|
|
301
|
+ |
approver_user_id: principal.user.id,
|
|
302
|
+ |
rule: "required_approvals",
|
|
303
|
+ |
request_digest: request.request_digest,
|
|
304
|
+ |
decided_at: DateTime.utc_now()
|
|
305
|
+ |
}
|
|
306
|
+ |
|> Approval.changeset(Map.put(attrs, "decision", decision))
|
|
307
|
+ |
|> Repo.insert()
|
|
308
|
+ |
|> case do
|
|
309
|
+ |
{:ok, _approval} ->
|
|
310
|
+ |
{:ok, _event} =
|
|
311
|
+ |
append_event(run, "approval_recorded", principal, %{"decision" => decision})
|
|
312
|
+ |
|
|
313
|
+ |
reevaluate(repository, environment, request, run, principal)
|
|
314
|
+ |
|
|
315
|
+ |
{:error, changeset} ->
|
|
316
|
+ |
{:error, changeset}
|
|
317
|
+ |
end
|
|
318
|
+ |
end
|
|
319
|
+ |
end
|
|
320
|
+ |
|
|
321
|
+ |
@doc "List the decisions recorded against one run."
|
|
322
|
+ |
@spec list_approvals(Repository.t(), Principal.t(), Run.t()) ::
|
|
323
|
+ |
{:ok, [Approval.t()]} | {:error, error()}
|
|
324
|
+ |
def list_approvals(%Repository{} = repository, %Principal{} = principal, %Run{} = run) do
|
|
325
|
+ |
with :ok <- Authority.authorize_read(principal, repository),
|
|
326
|
+ |
:ok <- Authority.check_repository(repository, run) do
|
|
327
|
+ |
{:ok,
|
|
328
|
+ |
Repo.all(
|
|
329
|
+ |
from approval in Approval,
|
|
330
|
+ |
where: approval.deployment_run_id == ^run.id,
|
|
331
|
+ |
order_by: [asc: approval.decided_at]
|
|
332
|
+ |
)}
|
|
333
|
+ |
end
|
|
334
|
+ |
end
|
|
335
|
+ |
|
|
336
|
+ |
# ---------------------------------------------------------------------------
|
|
337
|
+ |
# Check results
|
|
338
|
+ |
# ---------------------------------------------------------------------------
|
|
339
|
+ |
|
|
340
|
+ |
@doc """
|
|
341
|
+ |
Publish a check result for exact bytes, from a trusted identity.
|
|
342
|
+ |
|
|
343
|
+ |
Identity is `{repository, name, commit, artifact}`. Publishing the same name
|
|
344
|
+ |
for different bytes writes a new row, so a green result can never be replayed
|
|
345
|
+ |
onto an artifact it did not examine. Republishing the same bytes updates that
|
|
346
|
+ |
one row, which is how a `pending` check becomes `succeeded`.
|
|
347
|
+ |
|
|
348
|
+ |
Every run in `checking` for those bytes is re-evaluated afterwards, so evidence
|
|
349
|
+ |
arriving is what moves a run forward.
|
|
350
|
+ |
"""
|
|
351
|
+ |
@spec publish_check_result(Repository.t(), Principal.t(), map()) ::
|
|
352
|
+ |
{:ok, CheckResult.t()} | {:error, error()}
|
|
353
|
+ |
def publish_check_result(%Repository{} = repository, %Principal{} = principal, attrs) do
|
|
354
|
+ |
attrs = normalize(attrs)
|
|
355
|
+ |
|
|
356
|
+ |
with :ok <- Authority.authorize_publish_check(principal, repository, intent(attrs)) do
|
|
357
|
+ |
existing =
|
|
358
|
+ |
Repo.one(
|
|
359
|
+ |
from result in CheckResult,
|
|
360
|
+ |
where:
|
|
361
|
+ |
result.repository_id == ^repository.id and result.name == ^to_string(attrs["name"]) and
|
|
362
|
+ |
result.commit_sha == ^String.downcase(to_string(attrs["commit_sha"])) and
|
|
363
|
+ |
result.artifact_digest == ^to_string(attrs["artifact_digest"])
|
|
364
|
+ |
) || %CheckResult{repository_id: repository.id}
|
|
365
|
+ |
|
|
366
|
+ |
existing
|
|
367
|
+ |
|> CheckResult.changeset(attrs)
|
|
368
|
+ |
|> put_publisher(principal)
|
|
369
|
+ |
|> Repo.insert_or_update()
|
|
370
|
+ |
|> case do
|
|
371
|
+ |
{:ok, result} ->
|
|
372
|
+ |
reevaluate_checking_runs(repository, result, principal)
|
|
373
|
+ |
{:ok, result}
|
|
374
|
+ |
|
|
375
|
+ |
{:error, changeset} ->
|
|
376
|
+ |
{:error, changeset}
|
|
377
|
+ |
end
|
|
378
|
+ |
end
|
|
379
|
+ |
end
|
|
380
|
+ |
|
|
381
|
+ |
@doc "List the check results published for one repository's exact bytes."
|
|
382
|
+ |
@spec list_check_results(Repository.t(), Principal.t(), String.t(), String.t()) ::
|
|
383
|
+ |
{:ok, [CheckResult.t()]} | {:error, error()}
|
|
384
|
+ |
def list_check_results(
|
|
385
|
+ |
%Repository{} = repository,
|
|
386
|
+ |
%Principal{} = principal,
|
|
387
|
+ |
commit_sha,
|
|
388
|
+ |
artifact
|
|
389
|
+ |
) do
|
|
390
|
+ |
with :ok <- Authority.authorize_read(principal, repository) do
|
|
391
|
+ |
{:ok, check_results(repository, commit_sha, artifact)}
|
|
392
|
+ |
end
|
|
393
|
+ |
end
|
|
394
|
+ |
|
|
395
|
+ |
# ---------------------------------------------------------------------------
|
|
396
|
+ |
# Events
|
|
397
|
+ |
# ---------------------------------------------------------------------------
|
|
398
|
+ |
|
|
399
|
+ |
@doc """
|
|
400
|
+ |
List a run's events after a cursor, for polling or streaming.
|
|
401
|
+ |
|
|
402
|
+ |
The sequence is per-run and monotonic, so a reader that remembers the last
|
|
403
|
+ |
sequence it saw can resume without gaps or duplicates.
|
|
404
|
+ |
"""
|
|
405
|
+ |
@spec list_events(Repository.t(), Principal.t(), Run.t(), keyword()) ::
|
|
406
|
+ |
{:ok, [Event.t()]} | {:error, error()}
|
|
407
|
+ |
def list_events(
|
|
408
|
+ |
%Repository{} = repository,
|
|
409
|
+ |
%Principal{} = principal,
|
|
410
|
+ |
%Run{} = run,
|
|
411
|
+ |
options \\ []
|
|
412
|
+ |
) do
|
|
413
|
+ |
with :ok <- Authority.authorize_read(principal, repository),
|
|
414
|
+ |
:ok <- Authority.check_repository(repository, run) do
|
|
415
|
+ |
after_sequence = Keyword.get(options, :after_sequence, 0)
|
|
416
|
+ |
|
|
417
|
+ |
{:ok,
|
|
418
|
+ |
Repo.all(
|
|
419
|
+ |
from event in Event,
|
|
420
|
+ |
where: event.deployment_run_id == ^run.id and event.sequence > ^after_sequence,
|
|
421
|
+ |
order_by: [asc: event.sequence],
|
|
422
|
+ |
limit: ^limit(options)
|
|
423
|
+ |
)}
|
|
424
|
+ |
end
|
|
425
|
+ |
end
|
|
426
|
+ |
|
|
427
|
+ |
@doc "The topic a subscriber follows for one run's committed events."
|
|
428
|
+ |
@spec run_topic(Run.t() | String.t()) :: String.t()
|
|
429
|
+ |
def run_topic(%Run{id: id}), do: run_topic(id)
|
|
430
|
+ |
def run_topic(id) when is_binary(id), do: "deployments:run:" <> id
|
|
431
|
+ |
|
|
432
|
+ |
@doc """
|
|
433
|
+ |
Subscribe to one run's events.
|
|
434
|
+ |
|
|
435
|
+ |
Broadcasts happen after the transaction commits, so a subscriber never sees a
|
|
436
|
+ |
transition that later disappears.
|
|
437
|
+ |
"""
|
|
438
|
+ |
@spec subscribe(Run.t() | String.t()) :: :ok | {:error, term()}
|
|
439
|
+ |
def subscribe(run), do: Phoenix.PubSub.subscribe(OpenAgents.PubSub, run_topic(run))
|
|
440
|
+ |
|
|
441
|
+ |
# ---------------------------------------------------------------------------
|
|
442
|
+ |
# Workflow grants
|
|
443
|
+ |
# ---------------------------------------------------------------------------
|
|
444
|
+ |
|
|
445
|
+ |
@doc """
|
|
446
|
+ |
Issue a short-lived workflow grant bound to one run of one workflow.
|
|
447
|
+ |
|
|
448
|
+ |
The plaintext token is returned once and never stored. The grant carries the
|
|
449
|
+ |
repository, the optional environment, the source ref, the workflow, and the
|
|
450
|
+ |
workflow run it was issued for; authorization compares a request against those
|
|
451
|
+ |
bound values rather than against anything the caller sends.
|
|
452
|
+ |
"""
|
|
453
|
+ |
@spec issue_workflow_grant(Repository.t(), Principal.t(), map()) ::
|
|
454
|
+ |
{:ok, {WorkflowGrant.t(), String.t()}} | {:error, error()}
|
|
455
|
+ |
def issue_workflow_grant(%Repository{} = repository, %Principal{} = principal, attrs) do
|
|
456
|
+ |
attrs = normalize(attrs)
|
|
457
|
+ |
|
|
458
|
+ |
with :ok <- authorize_write(principal, repository),
|
|
459
|
+ |
{:ok, environment_id} <- grant_environment_id(repository, attrs["environment"]) do
|
|
460
|
+ |
plaintext = "oa_wfg_" <> Base.url_encode64(:crypto.strong_rand_bytes(32), padding: false)
|
|
461
|
+ |
lifetime = grant_lifetime(attrs["lifetime_seconds"])
|
|
462
|
+ |
|
|
463
|
+ |
%WorkflowGrant{
|
|
464
|
+ |
repository_id: repository.id,
|
|
465
|
+ |
environment_id: environment_id,
|
|
466
|
+ |
token_digest: digest(plaintext),
|
|
467
|
+ |
expires_at: DateTime.add(DateTime.utc_now(), lifetime, :second),
|
|
468
|
+ |
created_by_user_id: principal.user.id
|
|
469
|
+ |
}
|
|
470
|
+ |
|> WorkflowGrant.changeset(attrs)
|
|
471
|
+ |
|> Repo.insert()
|
|
472
|
+ |
|> case do
|
|
473
|
+ |
{:ok, grant} -> {:ok, {grant, plaintext}}
|
|
474
|
+ |
{:error, changeset} -> {:error, changeset}
|
|
475
|
+ |
end
|
|
476
|
+ |
end
|
|
477
|
+ |
end
|
|
478
|
+ |
|
|
479
|
+ |
@doc """
|
|
480
|
+ |
Authenticate a workflow grant token into a principal.
|
|
481
|
+ |
|
|
482
|
+ |
Only a live, unrevoked grant authenticates, and only the digest is compared, so
|
|
483
|
+ |
a database read cannot recover a usable token.
|
|
484
|
+ |
"""
|
|
485
|
+ |
@spec authenticate_workflow_grant(String.t()) ::
|
|
486
|
+ |
{:ok, Principal.t()} | {:error, :invalid_grant}
|
|
487
|
+ |
def authenticate_workflow_grant("oa_wfg_" <> _rest = plaintext)
|
|
488
|
+ |
when byte_size(plaintext) < 160 do
|
|
489
|
+ |
grant = Repo.one(from g in WorkflowGrant, where: g.token_digest == ^digest(plaintext))
|
|
490
|
+ |
|
|
491
|
+ |
if grant && WorkflowGrant.usable?(grant, DateTime.utc_now()) do
|
|
492
|
+ |
{:ok, Principal.workflow(grant)}
|
|
493
|
+ |
else
|
|
494
|
+ |
{:error, :invalid_grant}
|
|
495
|
+ |
end
|
|
496
|
+ |
end
|
|
497
|
+ |
|
|
498
|
+ |
def authenticate_workflow_grant(_plaintext), do: {:error, :invalid_grant}
|
|
499
|
+ |
|
|
500
|
+ |
@doc "Revoke a grant before it expires."
|
|
501
|
+ |
@spec revoke_workflow_grant(Repository.t(), Principal.t(), String.t()) ::
|
|
502
|
+ |
{:ok, WorkflowGrant.t()} | {:error, error()}
|
|
503
|
+ |
def revoke_workflow_grant(%Repository{} = repository, %Principal{} = principal, id) do
|
|
504
|
+ |
with :ok <- authorize_write(principal, repository),
|
|
505
|
+ |
%WorkflowGrant{} = grant <- Repo.get(WorkflowGrant, id),
|
|
506
|
+ |
:ok <- Authority.check_repository(repository, grant) do
|
|
507
|
+ |
grant
|
|
508
|
+ |
|> Ecto.Changeset.change(%{revoked_at: DateTime.utc_now()})
|
|
509
|
+ |
|> Repo.update()
|
|
510
|
+ |
else
|
|
511
|
+ |
nil -> {:error, :run_not_found}
|
|
512
|
+ |
{:error, reason} -> {:error, reason}
|
|
513
|
+ |
end
|
|
514
|
+ |
end
|
|
515
|
+ |
|
|
516
|
+ |
# ---------------------------------------------------------------------------
|
|
517
|
+ |
# Worker surface
|
|
518
|
+ |
# ---------------------------------------------------------------------------
|
|
519
|
+ |
|
|
520
|
+ |
@doc """
|
|
521
|
+ |
Claim the next queued run for execution, taking a lease.
|
|
522
|
+ |
|
|
523
|
+ |
The claim is a single conditional update, so two workers racing for one run
|
|
524
|
+ |
produce one winner and one `nil`: this is what keeps a provider from executing
|
|
525
|
+ |
the same run twice.
|
|
526
|
+ |
"""
|
|
527
|
+ |
@spec claim_run(String.t(), keyword()) :: {:ok, Run.t()} | :empty
|
|
528
|
+ |
def claim_run(worker, options \\ []) when is_binary(worker) do
|
|
529
|
+ |
lease_seconds = Keyword.get(options, :lease_seconds, @default_lease_seconds)
|
|
530
|
+ |
now = Keyword.get(options, :now, DateTime.utc_now())
|
|
531
|
+ |
expires_at = DateTime.add(now, lease_seconds, :second)
|
|
532
|
+ |
|
|
533
|
+ |
candidate =
|
|
534
|
+ |
Repo.one(
|
|
535
|
+ |
from run in Run,
|
|
536
|
+ |
where: run.state == "queued",
|
|
537
|
+ |
where: is_nil(run.lease_expires_at) or run.lease_expires_at < ^now,
|
|
538
|
+ |
order_by: [asc: run.inserted_at],
|
|
539
|
+ |
limit: 1,
|
|
540
|
+ |
select: run.id
|
|
541
|
+ |
)
|
|
542
|
+ |
|
|
543
|
+ |
case candidate && claim(candidate, worker, expires_at, now) do
|
|
544
|
+ |
nil ->
|
|
545
|
+ |
:empty
|
|
546
|
+ |
|
|
547
|
+ |
{1, [%Run{} = run]} ->
|
|
548
|
+ |
{:ok, _event} =
|
|
549
|
+ |
append_event(run, "lease_acquired", Principal.system(worker), %{"worker" => worker})
|
|
550
|
+ |
|
|
551
|
+ |
{:ok, run}
|
|
552
|
+ |
|
|
553
|
+ |
{0, _runs} ->
|
|
554
|
+ |
:empty
|
|
555
|
+ |
end
|
|
556
|
+ |
end
|
|
557
|
+ |
|
|
558
|
+ |
@doc """
|
|
559
|
+ |
Extend the lease a worker holds on a run.
|
|
560
|
+ |
|
|
561
|
+ |
A worker that cannot renew has lost the run, and must stop rather than keep
|
|
562
|
+ |
deploying: the renewal returning `:error` is the signal.
|
|
563
|
+ |
"""
|
|
564
|
+ |
@spec renew_lease(Run.t(), String.t(), keyword()) :: {:ok, Run.t()} | :error
|
|
565
|
+ |
def renew_lease(%Run{} = run, worker, options \\ []) do
|
|
566
|
+ |
lease_seconds = Keyword.get(options, :lease_seconds, @default_lease_seconds)
|
|
567
|
+ |
expires_at = DateTime.add(DateTime.utc_now(), lease_seconds, :second)
|
|
568
|
+ |
|
|
569
|
+ |
{count, runs} =
|
|
570
|
+ |
Repo.update_all(
|
|
571
|
+ |
from(r in Run,
|
|
572
|
+ |
where: r.id == ^run.id and r.lease_owner == ^worker and r.state == ^run.state,
|
|
573
|
+ |
select: r
|
|
574
|
+ |
),
|
|
575
|
+ |
set: [lease_expires_at: expires_at, updated_at: DateTime.utc_now()]
|
|
576
|
+ |
)
|
|
577
|
+ |
|
|
578
|
+ |
case {count, runs} do
|
|
579
|
+ |
{1, [%Run{} = renewed]} -> {:ok, renewed}
|
|
580
|
+ |
_lost -> :error
|
|
581
|
+ |
end
|
|
582
|
+ |
end
|
|
583
|
+ |
|
|
584
|
+ |
@doc """
|
|
585
|
+ |
Reclaim runs whose worker died holding a lease.
|
|
586
|
+ |
|
|
587
|
+ |
Recovery is deliberately narrow. A run that had not reached the provider goes
|
|
588
|
+ |
back to `queued` to be executed by another worker. A run that was already
|
|
589
|
+ |
`deploying` does *not*: the control plane does not know what the provider did,
|
|
590
|
+ |
so the run fails with an explicitly uncertain reason. Guessing `succeeded`
|
|
591
|
+ |
reports a deployment that may not exist, and guessing a clean retry can deploy
|
|
592
|
+ |
twice.
|
|
593
|
+ |
"""
|
|
594
|
+ |
@spec reconcile_leases(Principal.t(), keyword()) ::
|
|
595
|
+ |
{:ok, %{requeued: integer(), uncertain: integer()}} | {:error, error()}
|
|
596
|
+ |
def reconcile_leases(%Principal{} = principal, options \\ []) do
|
|
597
|
+ |
with :ok <- Authority.authorize_recovery(principal) do
|
|
598
|
+ |
now = Keyword.get(options, :now, DateTime.utc_now())
|
|
599
|
+ |
|
|
600
|
+ |
requeued = release_expired(now, principal)
|
|
601
|
+ |
uncertain = fail_uncertain(now, principal)
|
|
602
|
+ |
|
|
603
|
+ |
{:ok, %{requeued: requeued, uncertain: uncertain}}
|
|
604
|
+ |
end
|
|
605
|
+ |
end
|
|
606
|
+ |
|
|
607
|
+ |
@doc """
|
|
608
|
+ |
Record the provider's outcome for a run the worker holds.
|
|
609
|
+ |
|
|
610
|
+ |
Only three outcomes exist, and the third is why this function is narrow:
|
|
611
|
+ |
`{:uncertain, receipt}` writes `failed` with the reason
|
|
612
|
+ |
`provider_result_uncertain` rather than a success receipt. The control plane
|
|
613
|
+ |
never upgrades "I don't know" to "it worked".
|
|
614
|
+ |
"""
|
|
615
|
+ |
@spec finish_run(Run.t(), String.t(), Provider.result(), keyword()) ::
|
|
616
|
+ |
{:ok, Run.t()} | {:error, error()}
|
|
617
|
+ |
def finish_run(%Run{} = run, worker, result, options \\ []) do
|
|
618
|
+ |
principal = Principal.system(worker)
|
|
619
|
+ |
secret_values = Keyword.get(options, :secret_values, [])
|
|
620
|
+ |
|
|
621
|
+ |
# A cancellation requested while the provider was running is honored here,
|
|
622
|
+ |
# after the provider reported, so the run's terminal state reflects what the
|
|
623
|
+ |
# provider actually did rather than what the canceller hoped.
|
|
624
|
+ |
cancelled? = not is_nil(run.cancel_requested_at)
|
|
625
|
+ |
|
|
626
|
+ |
case result do
|
|
627
|
+ |
{:ok, receipt} when cancelled? ->
|
|
628
|
+ |
transition(run, "cancelled", principal,
|
|
629
|
+ |
reason: "cancelled_after_provider_success",
|
|
630
|
+ |
receipt: receipt,
|
|
631
|
+ |
secret_values: secret_values,
|
|
632
|
+ |
finished: true
|
|
633
|
+ |
)
|
|
634
|
+ |
|
|
635
|
+ |
{:ok, receipt} ->
|
|
636
|
+ |
transition(run, "succeeded", principal,
|
|
637
|
+ |
receipt: receipt,
|
|
638
|
+ |
secret_values: secret_values,
|
|
639
|
+ |
finished: true
|
|
640
|
+ |
)
|
|
641
|
+ |
|
|
642
|
+ |
{:error, reason} when cancelled? ->
|
|
643
|
+ |
transition(run, "cancelled", principal,
|
|
644
|
+ |
reason: bounded_reason(reason),
|
|
645
|
+ |
finished: true
|
|
646
|
+ |
)
|
|
647
|
+ |
|
|
648
|
+ |
{:error, reason} ->
|
|
649
|
+ |
transition(run, "failed", principal,
|
|
650
|
+ |
reason: bounded_reason(reason),
|
|
651
|
+ |
finished: true
|
|
652
|
+ |
)
|
|
653
|
+ |
|
|
654
|
+ |
{:uncertain, receipt} ->
|
|
655
|
+ |
transition(run, "failed", principal,
|
|
656
|
+ |
reason: "provider_result_uncertain",
|
|
657
|
+ |
receipt: receipt,
|
|
658
|
+ |
secret_values: secret_values,
|
|
659
|
+ |
finished: true
|
|
660
|
+ |
)
|
|
661
|
+ |
end
|
|
662
|
+ |
end
|
|
663
|
+ |
|
|
664
|
+ |
@doc """
|
|
665
|
+ |
Move a run to `state`, transactionally, with its event.
|
|
666
|
+ |
|
|
667
|
+ |
This is the only writer of `Run.state`. It checks the transition against
|
|
668
|
+ |
`OpenAgents.Deployments.Lifecycle`, compares the state it read against the row
|
|
669
|
+ |
it updates so a concurrent transition cannot be lost, and appends the event in
|
|
670
|
+ |
the same transaction. A state without a matching event, or an event without a
|
|
671
|
+ |
state change, would each make the history a guess.
|
|
672
|
+ |
"""
|
|
673
|
+ |
@spec transition(Run.t(), String.t(), Principal.t(), keyword()) ::
|
|
674
|
+ |
{:ok, Run.t()} | {:error, error()}
|
|
675
|
+ |
def transition(%Run{} = run, state, %Principal{} = principal, options \\ []) do
|
|
676
|
+ |
with :ok <- Lifecycle.check(run.state, state) do
|
|
677
|
+ |
now = DateTime.utc_now()
|
|
678
|
+ |
finished = Keyword.get(options, :finished, state in Run.terminal_states())
|
|
679
|
+ |
|
|
680
|
+ |
changes =
|
|
681
|
+ |
%{state: state, updated_at: now}
|
|
682
|
+ |
|> maybe_put(:result_reason, Keyword.get(options, :reason))
|
|
683
|
+ |
|> maybe_put(
|
|
684
|
+ |
:provider_receipt,
|
|
685
|
+ |
case Keyword.get(options, :receipt) do
|
|
686
|
+ |
nil ->
|
|
687
|
+ |
nil
|
|
688
|
+ |
|
|
689
|
+ |
receipt ->
|
|
690
|
+ |
Provider.sanitize_receipt(receipt, Keyword.get(options, :secret_values, []))
|
|
691
|
+ |
end
|
|
692
|
+ |
)
|
|
693
|
+ |
|> maybe_put(:policy_explanation, Keyword.get(options, :explanation))
|
|
694
|
+ |
|> maybe_put(:superseded_by_run_id, Keyword.get(options, :superseded_by_run_id))
|
|
695
|
+ |
|> maybe_put(:started_at, if(state == "deploying", do: now))
|
|
696
|
+ |
|> maybe_put(:finished_at, if(finished, do: now))
|
|
697
|
+ |
|> maybe_put(:lease_owner, if(finished, do: :nil_value))
|
|
698
|
+ |
|> maybe_put(:lease_expires_at, if(finished, do: :nil_value))
|
|
699
|
+ |
|
|
700
|
+ |
Multi.new()
|
|
701
|
+ |
|> Multi.run(:run, fn repo, _changes ->
|
|
702
|
+ |
{count, runs} =
|
|
703
|
+ |
repo.update_all(
|
|
704
|
+ |
from(r in Run, where: r.id == ^run.id and r.state == ^run.state, select: r),
|
|
705
|
+ |
set: Enum.map(changes, fn {key, value} -> {key, denil(value)} end)
|
|
706
|
+ |
)
|
|
707
|
+ |
|
|
708
|
+ |
case {count, runs} do
|
|
709
|
+ |
{1, [%Run{} = updated]} -> {:ok, updated}
|
|
710
|
+ |
_stale -> {:error, :precondition_failed}
|
|
711
|
+ |
end
|
|
712
|
+ |
end)
|
|
713
|
+ |
|> Multi.run(:event, fn repo, %{run: updated} ->
|
|
714
|
+ |
insert_event(
|
|
715
|
+ |
repo,
|
|
716
|
+ |
updated,
|
|
717
|
+ |
Keyword.get(options, :event_type, "state_changed"),
|
|
718
|
+ |
principal,
|
|
719
|
+ |
Map.merge(
|
|
720
|
+ |
%{"reason" => Keyword.get(options, :reason)},
|
|
721
|
+ |
Keyword.get(options, :detail, %{})
|
|
722
|
+ |
),
|
|
723
|
+ |
from: run.state,
|
|
724
|
+ |
to: state
|
|
725
|
+ |
)
|
|
726
|
+ |
end)
|
|
727
|
+ |
|> Repo.transaction()
|
|
728
|
+ |
|> case do
|
|
729
|
+ |
{:ok, %{run: updated, event: event}} ->
|
|
730
|
+ |
broadcast(updated, event)
|
|
731
|
+ |
{:ok, updated}
|
|
732
|
+ |
|
|
733
|
+ |
{:error, _step, reason, _changes} ->
|
|
734
|
+ |
{:error, reason}
|
|
735
|
+ |
end
|
|
736
|
+ |
end
|
|
737
|
+ |
end
|
|
738
|
+ |
|
|
739
|
+ |
@doc """
|
|
740
|
+ |
Re-evaluate policy for a run that is waiting, re-checking authority first.
|
|
741
|
+ |
|
|
742
|
+ |
Called when evidence arrives, when a decision is recorded, and before a worker
|
|
743
|
+ |
hands a run to a provider. A run admitted an hour ago is not admitted now if
|
|
744
|
+ |
the environment froze, the window closed, or the requester lost membership.
|
|
745
|
+ |
"""
|
|
746
|
+ |
@spec reevaluate(Repository.t(), Environment.t(), Request.t(), Run.t(), Principal.t()) ::
|
|
747
|
+ |
{:ok, Run.t()} | {:error, error()}
|
|
748
|
+ |
def reevaluate(
|
|
749
|
+ |
%Repository{} = repository,
|
|
750
|
+ |
%Environment{} = environment,
|
|
751
|
+ |
%Request{} = request,
|
|
752
|
+ |
%Run{} = run,
|
|
753
|
+ |
%Principal{} = principal
|
|
754
|
+ |
) do
|
|
755
|
+ |
approvals =
|
|
756
|
+ |
Repo.all(from approval in Approval, where: approval.deployment_run_id == ^run.id)
|
|
757
|
+ |
|
|
758
|
+ |
checks = check_results(repository, request.commit_sha, request.artifact_digest)
|
|
759
|
+ |
|
|
760
|
+ |
case Policy.evaluate(environment, request, checks, approvals, DateTime.utc_now()) do
|
|
761
|
+ |
{:admit, state, explanation} ->
|
|
762
|
+ |
if state == run.state do
|
|
763
|
+ |
{:ok, store_explanation(run, explanation)}
|
|
764
|
+ |
else
|
|
765
|
+ |
transition(run, state, principal,
|
|
766
|
+ |
explanation: explanation,
|
|
767
|
+ |
event_type: "policy_evaluated"
|
|
768
|
+ |
)
|
|
769
|
+ |
end
|
|
770
|
+ |
|
|
771
|
+ |
{:deny, reason, explanation} ->
|
|
772
|
+ |
transition(run, "failed", principal,
|
|
773
|
+ |
reason: reason,
|
|
774
|
+ |
explanation: explanation,
|
|
775
|
+ |
event_type: "policy_denied"
|
|
776
|
+ |
)
|
|
777
|
+ |
end
|
|
778
|
+ |
end
|
|
779
|
+ |
|
|
780
|
+ |
@doc """
|
|
781
|
+ |
Re-check the authority behind a request at a sensitive transition.
|
|
782
|
+ |
|
|
783
|
+ |
Membership is revoked, roles change, and grants expire between `requested` and
|
|
784
|
+ |
`deploying`. The check reruns against the durable request rather than the
|
|
785
|
+ |
original credential, because the original credential is long gone by then.
|
|
786
|
+ |
"""
|
|
787
|
+ |
@spec recheck_request_authority(Repository.t(), Environment.t(), Request.t()) ::
|
|
788
|
+ |
:ok | {:error, error()}
|
|
789
|
+ |
def recheck_request_authority(
|
|
790
|
+ |
%Repository{} = repository,
|
|
791
|
+ |
%Environment{} = environment,
|
|
792
|
+ |
%Request{} = request
|
|
793
|
+ |
) do
|
|
794
|
+ |
case request.principal_type do
|
|
795
|
+ |
"user" ->
|
|
796
|
+ |
case Repo.get(User, request.requested_by_user_id) do
|
|
797
|
+ |
%User{} = user ->
|
|
798
|
+ |
Authority.authorize_request(Principal.user(user), repository, environment, %{})
|
|
799
|
+ |
|
|
800
|
+ |
nil ->
|
|
801
|
+ |
{:error, {:forbidden, :not_a_member}}
|
|
802
|
+ |
end
|
|
803
|
+ |
|
|
804
|
+ |
"workflow" ->
|
|
805
|
+ |
case Repo.get(WorkflowGrant, request.requested_by_grant_id) do
|
|
806
|
+ |
%WorkflowGrant{} = grant ->
|
|
807
|
+ |
Authority.authorize_request(
|
|
808
|
+ |
Principal.workflow(grant),
|
|
809
|
+ |
repository,
|
|
810
|
+ |
environment,
|
|
811
|
+ |
%{}
|
|
812
|
+ |
)
|
|
813
|
+ |
|
|
814
|
+ |
nil ->
|
|
815
|
+ |
{:error, {:forbidden, :grant_expired}}
|
|
816
|
+ |
end
|
|
817
|
+ |
|
|
818
|
+ |
_other ->
|
|
819
|
+ |
{:error, {:forbidden, :operator_is_not_tenant}}
|
|
820
|
+ |
end
|
|
821
|
+ |
end
|
|
822
|
+ |
|
|
823
|
+ |
@doc "Build the execution a provider receives, resolving secrets for this attempt."
|
|
824
|
+ |
@spec build_execution(Repository.t(), Environment.t(), Request.t(), Run.t(), keyword()) ::
|
|
825
|
+ |
{:ok, OpenAgents.Deployments.Execution.t()} | {:error, error()}
|
|
826
|
+ |
def build_execution(
|
|
827
|
+ |
%Repository{} = repository,
|
|
828
|
+ |
%Environment{} = environment,
|
|
829
|
+ |
%Request{} = request,
|
|
830
|
+ |
%Run{} = run,
|
|
831
|
+ |
options \\ []
|
|
832
|
+ |
) do
|
|
833
|
+ |
with {:ok, module} <- Provider.fetch(environment.provider),
|
|
834
|
+ |
references = module.required_secret_references(environment.provider_config),
|
|
835
|
+ |
{:ok, secrets} <- OpenAgents.Deployments.SecretResolver.resolve(environment, references) do
|
|
836
|
+ |
{:ok,
|
|
837
|
+ |
%OpenAgents.Deployments.Execution{
|
|
838
|
+ |
run_id: run.id,
|
|
839
|
+ |
repository: repository.owner <> "/" <> repository.name,
|
|
840
|
+ |
environment: environment.name,
|
|
841
|
+ |
commit_sha: request.commit_sha,
|
|
842
|
+ |
artifact_digest: request.artifact_digest,
|
|
843
|
+ |
input_digest: run.input_digest,
|
|
844
|
+ |
attempt: run.attempt_count + 1,
|
|
845
|
+ |
deadline:
|
|
846
|
+ |
DateTime.add(DateTime.utc_now(), Keyword.get(options, :timeout_seconds, 600), :second),
|
|
847
|
+ |
provider_config: environment.provider_config,
|
|
848
|
+ |
secrets: secrets
|
|
849
|
+ |
}}
|
|
850
|
+ |
end
|
|
851
|
+ |
end
|
|
852
|
+ |
|
|
853
|
+ |
@doc "Load the repository, environment, and request a run belongs to."
|
|
854
|
+ |
@spec load_run_context(Run.t()) ::
|
|
855
|
+ |
{:ok, {Repository.t(), Environment.t(), Request.t()}} | {:error, error()}
|
|
856
|
+ |
def load_run_context(%Run{} = run) do
|
|
857
|
+ |
with %Repository{} = repository <- Repo.get(Repository, run.repository_id),
|
|
858
|
+ |
%Environment{} = environment <- Repo.get(Environment, run.environment_id),
|
|
859
|
+ |
{:ok, request} <- fetch_request_record(run.deployment_request_id) do
|
|
860
|
+ |
{:ok, {repository, environment, request}}
|
|
861
|
+ |
else
|
|
862
|
+ |
nil -> {:error, :run_not_found}
|
|
863
|
+ |
{:error, reason} -> {:error, reason}
|
|
864
|
+ |
end
|
|
865
|
+ |
end
|
|
866
|
+ |
|
|
867
|
+ |
# ---------------------------------------------------------------------------
|
|
868
|
+ |
# Admission
|
|
869
|
+ |
# ---------------------------------------------------------------------------
|
|
870
|
+ |
|
|
871
|
+ |
defp admit(repository, environment, request, principal, _provider) do
|
|
872
|
+ |
protection = environment.protection || %Protection{}
|
|
873
|
+ |
|
|
874
|
+ |
with {:ok, run} <- insert_run(repository, environment, request),
|
|
875
|
+ |
{:ok, run} <- apply_concurrency(repository, environment, protection, run, principal) do
|
|
876
|
+ |
reevaluate(repository, environment, request, run, principal)
|
|
877
|
+ |
end
|
|
878
|
+ |
end
|
|
879
|
+ |
|
|
880
|
+ |
defp insert_run(repository, environment, request) do
|
|
881
|
+ |
run = %Run{
|
|
882
|
+ |
repository_id: repository.id,
|
|
883
|
+ |
environment_id: environment.id,
|
|
884
|
+ |
deployment_request_id: request.id,
|
|
885
|
+ |
input_digest: request.input_digest,
|
|
886
|
+ |
provider: environment.provider
|
|
887
|
+ |
}
|
|
888
|
+ |
|
|
889
|
+ |
Multi.new()
|
|
890
|
+ |
|> Multi.insert(
|
|
891
|
+ |
:run,
|
|
892
|
+ |
Run.changeset(run, %{state: Policy.initial_state(environment)})
|
|
893
|
+ |
)
|
|
894
|
+ |
|> Multi.run(:event, fn repo, %{run: inserted} ->
|
|
895
|
+ |
insert_event(repo, inserted, "run_created", Principal.system("control_plane"), %{},
|
|
896
|
+ |
to: inserted.state
|
|
897
|
+ |
)
|
|
898
|
+ |
end)
|
|
899
|
+ |
|> Repo.transaction()
|
|
900
|
+ |
|> case do
|
|
901
|
+ |
{:ok, %{run: run, event: event}} ->
|
|
902
|
+ |
broadcast(run, event)
|
|
903
|
+ |
{:ok, run}
|
|
904
|
+ |
|
|
905
|
+ |
{:error, _step, reason, _changes} ->
|
|
906
|
+ |
{:error, reason}
|
|
907
|
+ |
end
|
|
908
|
+ |
end
|
|
909
|
+ |
|
|
910
|
+ |
# Concurrency is per environment and never crosses environments: superseding a
|
|
911
|
+ |
# preview deployment must not supersede production, so candidates are selected
|
|
912
|
+ |
# by environment id alone.
|
|
913
|
+ |
defp apply_concurrency(
|
|
914
|
+ |
_repository,
|
|
915
|
+ |
_environment,
|
|
916
|
+ |
%Protection{concurrency: "queue"},
|
|
917
|
+ |
run,
|
|
918
|
+ |
_principal
|
|
919
|
+ |
),
|
|
920
|
+ |
do: {:ok, run}
|
|
921
|
+ |
|
|
922
|
+ |
defp apply_concurrency(_repository, environment, %Protection{concurrency: mode}, run, principal) do
|
|
923
|
+ |
others =
|
|
924
|
+ |
Repo.all(
|
|
925
|
+ |
from other in Run,
|
|
926
|
+ |
where:
|
|
927
|
+ |
other.environment_id == ^environment.id and other.id != ^run.id and
|
|
928
|
+ |
other.state in ^Run.active_states(),
|
|
929
|
+ |
order_by: [asc: other.inserted_at]
|
|
930
|
+ |
)
|
|
931
|
+ |
|
|
932
|
+ |
case mode do
|
|
933
|
+ |
"reject" when others != [] ->
|
|
934
|
+ |
transition(run, "failed", principal,
|
|
935
|
+ |
reason: "environment_busy",
|
|
936
|
+ |
event_type: "policy_denied"
|
|
937
|
+ |
)
|
|
938
|
+ |
|
|
939
|
+ |
"cancel" ->
|
|
940
|
+ |
for other <- others, other.state != "deploying" do
|
|
941
|
+ |
transition(other, "cancelled", principal, reason: "superseded_by_newer_request")
|
|
942
|
+ |
end
|
|
943
|
+ |
|
|
944
|
+ |
{:ok, run}
|
|
945
|
+ |
|
|
946
|
+ |
"supersede" ->
|
|
947
|
+ |
for other <- others, other.state != "deploying" do
|
|
948
|
+ |
transition(other, "superseded", principal,
|
|
949
|
+ |
reason: "superseded_by_newer_request",
|
|
950
|
+ |
superseded_by_run_id: run.id
|
|
951
|
+ |
)
|
|
952
|
+ |
end
|
|
953
|
+ |
|
|
954
|
+ |
{:ok, run}
|
|
955
|
+ |
|
|
956
|
+ |
_queue_or_idle ->
|
|
957
|
+ |
{:ok, run}
|
|
958
|
+ |
end
|
|
959
|
+ |
end
|
|
960
|
+ |
|
|
961
|
+ |
defp insert_request(repository, environment, principal, principal_type, attrs) do
|
|
962
|
+ |
input_digest =
|
|
963
|
+ |
digest([repository.id, environment.id, attrs["commit_sha"], attrs["artifact_digest"]])
|
|
964
|
+ |
|
|
965
|
+ |
request_digest =
|
|
966
|
+ |
digest([input_digest, attrs["source_ref"], attrs["source_workflow"] || ""])
|
|
967
|
+ |
|
|
968
|
+ |
request = %Request{
|
|
969
|
+ |
repository_id: repository.id,
|
|
970
|
+ |
environment_id: environment.id,
|
|
971
|
+ |
principal_type: principal_type,
|
|
972
|
+ |
requested_by_user_id: principal.user && principal.user.id,
|
|
973
|
+ |
requested_by_grant_id: principal.grant && principal.grant.id,
|
|
974
|
+ |
request_digest: request_digest,
|
|
975
|
+ |
input_digest: input_digest,
|
|
976
|
+ |
requested_at: DateTime.utc_now()
|
|
977
|
+ |
}
|
|
978
|
+ |
|
|
979
|
+ |
case Repo.insert(Request.changeset(request, attrs)) do
|
|
980
|
+ |
{:ok, inserted} ->
|
|
981
|
+ |
{:ok, inserted}
|
|
982
|
+ |
|
|
983
|
+ |
{:error, changeset} ->
|
|
984
|
+ |
replay(repository, environment, attrs, request_digest, changeset)
|
|
985
|
+ |
end
|
|
986
|
+ |
end
|
|
987
|
+ |
|
|
988
|
+ |
# An idempotency key is a promise about bytes. Replaying it with the same
|
|
989
|
+ |
# request digest returns the original record; replaying it with anything else
|
|
990
|
+ |
# is a conflict, because the caller believes it deployed something it did not.
|
|
991
|
+ |
defp replay(repository, environment, attrs, request_digest, changeset) do
|
|
992
|
+ |
existing =
|
|
993
|
+ |
Repo.one(
|
|
994
|
+ |
from request in Request,
|
|
995
|
+ |
where:
|
|
996
|
+ |
request.repository_id == ^repository.id and
|
|
997
|
+ |
request.environment_id == ^environment.id and
|
|
998
|
+ |
request.idempotency_key == ^to_string(attrs["idempotency_key"])
|
|
999
|
+ |
)
|
|
1000
|
+ |
|
|
1001
|
+ |
cond do
|
|
1002
|
+ |
is_nil(existing) -> {:error, changeset}
|
|
1003
|
+ |
existing.request_digest == request_digest -> {:ok, existing}
|
|
1004
|
+ |
true -> {:error, :idempotency_conflict}
|
|
1005
|
+ |
end
|
|
1006
|
+ |
end
|
|
1007
|
+ |
|
|
1008
|
+ |
defp reevaluate_checking_runs(repository, %CheckResult{} = result, _principal) do
|
|
1009
|
+ |
runs =
|
|
1010
|
+ |
Repo.all(
|
|
1011
|
+ |
from run in Run,
|
|
1012
|
+ |
join: request in Request,
|
|
1013
|
+ |
on: request.id == run.deployment_request_id,
|
|
1014
|
+ |
where:
|
|
1015
|
+ |
run.repository_id == ^repository.id and run.state in ~w(requested checking) and
|
|
1016
|
+ |
request.commit_sha == ^result.commit_sha and
|
|
1017
|
+ |
request.artifact_digest == ^result.artifact_digest,
|
|
1018
|
+ |
preload: [:environment, :deployment_request]
|
|
1019
|
+ |
)
|
|
1020
|
+ |
|
|
1021
|
+ |
for run <- runs do
|
|
1022
|
+ |
reevaluate(
|
|
1023
|
+ |
repository,
|
|
1024
|
+ |
run.environment,
|
|
1025
|
+ |
run.deployment_request,
|
|
1026
|
+ |
run,
|
|
1027
|
+ |
Principal.system("control_plane")
|
|
1028
|
+ |
)
|
|
1029
|
+ |
end
|
|
1030
|
+ |
|
|
1031
|
+ |
:ok
|
|
1032
|
+ |
end
|
|
1033
|
+ |
|
|
1034
|
+ |
# ---------------------------------------------------------------------------
|
|
1035
|
+ |
# Events
|
|
1036
|
+ |
# ---------------------------------------------------------------------------
|
|
1037
|
+ |
|
|
1038
|
+ |
defp append_event(%Run{} = run, type, principal, detail) do
|
|
1039
|
+ |
case insert_event(Repo, run, type, principal, detail, []) do
|
|
1040
|
+ |
{:ok, event} ->
|
|
1041
|
+ |
broadcast(run, event)
|
|
1042
|
+ |
{:ok, event}
|
|
1043
|
+ |
|
|
1044
|
+ |
{:error, reason} ->
|
|
1045
|
+ |
{:error, reason}
|
|
1046
|
+ |
end
|
|
1047
|
+ |
end
|
|
1048
|
+ |
|
|
1049
|
+ |
defp insert_event(repo, %Run{} = run, type, %Principal{} = principal, detail, states) do
|
|
1050
|
+ |
sequence =
|
|
1051
|
+ |
repo.one(
|
|
1052
|
+ |
from event in Event,
|
|
1053
|
+ |
where: event.deployment_run_id == ^run.id,
|
|
1054
|
+ |
select: coalesce(max(event.sequence), 0)
|
|
1055
|
+ |
) + 1
|
|
1056
|
+ |
|
|
1057
|
+ |
%Event{
|
|
1058
|
+ |
repository_id: run.repository_id,
|
|
1059
|
+ |
deployment_run_id: run.id,
|
|
1060
|
+ |
actor_type: Principal.actor_type(principal),
|
|
1061
|
+ |
actor_id: Principal.actor_id(principal)
|
|
1062
|
+ |
}
|
|
1063
|
+ |
|> Event.changeset(%{
|
|
1064
|
+ |
"sequence" => sequence,
|
|
1065
|
+ |
"type" => type,
|
|
1066
|
+ |
"from_state" => Keyword.get(states, :from),
|
|
1067
|
+ |
"to_state" => Keyword.get(states, :to),
|
|
1068
|
+ |
"detail" => scrub(detail)
|
|
1069
|
+ |
})
|
|
1070
|
+ |
|> repo.insert()
|
|
1071
|
+ |
end
|
|
1072
|
+ |
|
|
1073
|
+ |
# Event detail reaches tenant reads and audit exports, so it is redacted and
|
|
1074
|
+ |
# bounded here rather than at each caller.
|
|
1075
|
+ |
defp scrub(detail) when is_map(detail) do
|
|
1076
|
+ |
detail
|
|
1077
|
+ |
|> Enum.reject(fn {_key, value} -> is_nil(value) end)
|
|
1078
|
+ |
|> Map.new(fn {key, value} -> {to_string(key), scrub_value(value)} end)
|
|
1079
|
+ |
end
|
|
1080
|
+ |
|
|
1081
|
+ |
defp scrub_value(value) when is_binary(value),
|
|
1082
|
+ |
do: value |> OpenAgents.LogSafety.redact() |> String.slice(0, 500)
|
|
1083
|
+ |
|
|
1084
|
+ |
defp scrub_value(value) when is_list(value), do: Enum.map(value, &scrub_value/1)
|
|
1085
|
+ |
defp scrub_value(value) when is_map(value), do: scrub(value)
|
|
1086
|
+ |
defp scrub_value(value), do: value
|
|
1087
|
+ |
|
|
1088
|
+ |
defp broadcast(%Run{} = run, %Event{} = event) do
|
|
1089
|
+ |
Phoenix.PubSub.broadcast(
|
|
1090
|
+ |
OpenAgents.PubSub,
|
|
1091
|
+ |
run_topic(run),
|
|
1092
|
+ |
{:deployment_event,
|
|
1093
|
+ |
%{run_id: run.id, sequence: event.sequence, type: event.type, state: run.state}}
|
|
1094
|
+ |
)
|
|
1095
|
+ |
end
|
|
1096
|
+ |
|
|
1097
|
+ |
# ---------------------------------------------------------------------------
|
|
1098
|
+ |
# Recovery
|
|
1099
|
+ |
# ---------------------------------------------------------------------------
|
|
1100
|
+ |
|
|
1101
|
+ |
defp release_expired(now, principal) do
|
|
1102
|
+ |
runs =
|
|
1103
|
+ |
Repo.all(
|
|
1104
|
+ |
from run in Run,
|
|
1105
|
+ |
where:
|
|
1106
|
+ |
run.state == "queued" and not is_nil(run.lease_expires_at) and
|
|
1107
|
+ |
run.lease_expires_at < ^now
|
|
1108
|
+ |
)
|
|
1109
|
+ |
|
|
1110
|
+ |
for run <- runs do
|
|
1111
|
+ |
Repo.update_all(
|
|
1112
|
+ |
from(r in Run, where: r.id == ^run.id),
|
|
1113
|
+ |
set: [lease_owner: nil, lease_expires_at: nil, updated_at: DateTime.utc_now()]
|
|
1114
|
+ |
)
|
|
1115
|
+ |
|
|
1116
|
+ |
append_event(run, "lease_released", principal, %{"reason" => "lease_expired"})
|
|
1117
|
+ |
end
|
|
1118
|
+ |
|
|
1119
|
+ |
length(runs)
|
|
1120
|
+ |
end
|
|
1121
|
+ |
|
|
1122
|
+ |
defp fail_uncertain(now, principal) do
|
|
1123
|
+ |
runs =
|
|
1124
|
+ |
Repo.all(
|
|
1125
|
+ |
from run in Run,
|
|
1126
|
+ |
where:
|
|
1127
|
+ |
run.state == "deploying" and not is_nil(run.lease_expires_at) and
|
|
1128
|
+ |
run.lease_expires_at < ^now
|
|
1129
|
+ |
)
|
|
1130
|
+ |
|
|
1131
|
+ |
for run <- runs do
|
|
1132
|
+ |
transition(run, "failed", principal,
|
|
1133
|
+ |
reason: "provider_result_uncertain",
|
|
1134
|
+ |
event_type: "recovered",
|
|
1135
|
+ |
detail: %{"recovery" => "lease_expired_while_deploying"}
|
|
1136
|
+ |
)
|
|
1137
|
+ |
end
|
|
1138
|
+ |
|
|
1139
|
+ |
length(runs)
|
|
1140
|
+ |
end
|
|
1141
|
+ |
|
|
1142
|
+ |
# ---------------------------------------------------------------------------
|
|
1143
|
+ |
# Helpers
|
|
1144
|
+ |
# ---------------------------------------------------------------------------
|
|
1145
|
+ |
|
|
1146
|
+ |
defp authorize_write(%Principal{kind: :user, user: %User{} = user}, %Repository{} = repository) do
|
|
1147
|
+ |
if OpenAgents.Repositories.writable?(repository, user) do
|
|
1148
|
+ |
:ok
|
|
1149
|
+ |
else
|
|
1150
|
+ |
{:error, {:forbidden, :not_writable}}
|
|
1151
|
+ |
end
|
|
1152
|
+ |
end
|
|
1153
|
+ |
|
|
1154
|
+ |
defp authorize_write(%Principal{}, %Repository{}), do: {:error, {:forbidden, :not_writable}}
|
|
1155
|
+ |
|
|
1156
|
+ |
defp fetch_environment_record(%Repository{} = repository, name) when is_binary(name) do
|
|
1157
|
+ |
case Repo.one(
|
|
1158
|
+ |
from environment in Environment,
|
|
1159
|
+ |
where: environment.repository_id == ^repository.id and environment.name == ^name
|
|
1160
|
+ |
) do
|
|
1161
|
+ |
%Environment{} = environment -> {:ok, environment}
|
|
1162
|
+ |
nil -> {:error, :environment_not_found}
|
|
1163
|
+ |
end
|
|
1164
|
+ |
end
|
|
1165
|
+ |
|
|
1166
|
+ |
defp fetch_environment_record(%Repository{}, _name), do: {:error, :environment_not_found}
|
|
1167
|
+ |
|
|
1168
|
+ |
defp fetch_environment_by_id(%Repository{} = repository, id) do
|
|
1169
|
+ |
case Repo.get(Environment, id) do
|
|
1170
|
+ |
%Environment{} = environment ->
|
|
1171
|
+ |
with :ok <- Authority.check_repository(repository, environment), do: {:ok, environment}
|
|
1172
|
+ |
|
|
1173
|
+ |
nil ->
|
|
1174
|
+ |
{:error, :environment_not_found}
|
|
1175
|
+ |
end
|
|
1176
|
+ |
end
|
|
1177
|
+ |
|
|
1178
|
+ |
defp fetch_run_record(id) when is_binary(id) do
|
|
1179
|
+ |
case Ecto.UUID.cast(id) do
|
|
1180
|
+ |
{:ok, uuid} ->
|
|
1181
|
+ |
case Repo.get(Run, uuid) do
|
|
1182
|
+ |
%Run{} = run -> {:ok, run}
|
|
1183
|
+ |
nil -> {:error, :run_not_found}
|
|
1184
|
+ |
end
|
|
1185
|
+ |
|
|
1186
|
+ |
:error ->
|
|
1187
|
+ |
{:error, :run_not_found}
|
|
1188
|
+ |
end
|
|
1189
|
+ |
end
|
|
1190
|
+ |
|
|
1191
|
+ |
defp fetch_run_record(_id), do: {:error, :run_not_found}
|
|
1192
|
+ |
|
|
1193
|
+ |
defp fetch_request_record(id) do
|
|
1194
|
+ |
case Repo.get(Request, id) do
|
|
1195
|
+ |
%Request{} = request -> {:ok, request}
|
|
1196
|
+ |
nil -> {:error, :request_not_found}
|
|
1197
|
+ |
end
|
|
1198
|
+ |
end
|
|
1199
|
+ |
|
|
1200
|
+ |
defp check_results(%Repository{} = repository, commit_sha, artifact_digest) do
|
|
1201
|
+ |
Repo.all(
|
|
1202
|
+ |
from result in CheckResult,
|
|
1203
|
+ |
where:
|
|
1204
|
+ |
result.repository_id == ^repository.id and result.commit_sha == ^commit_sha and
|
|
1205
|
+ |
result.artifact_digest == ^artifact_digest
|
|
1206
|
+ |
)
|
|
1207
|
+ |
end
|
|
1208
|
+ |
|
|
1209
|
+ |
defp store_explanation(%Run{} = run, explanation) do
|
|
1210
|
+ |
{_count, _runs} =
|
|
1211
|
+ |
Repo.update_all(
|
|
1212
|
+ |
from(r in Run, where: r.id == ^run.id),
|
|
1213
|
+ |
set: [policy_explanation: explanation, updated_at: DateTime.utc_now()]
|
|
1214
|
+ |
)
|
|
1215
|
+ |
|
|
1216
|
+ |
%Run{run | policy_explanation: explanation}
|
|
1217
|
+ |
end
|
|
1218
|
+ |
|
|
1219
|
+ |
defp claim(run_id, worker, expires_at, now) do
|
|
1220
|
+ |
Repo.update_all(
|
|
1221
|
+ |
from(run in Run,
|
|
1222
|
+ |
where: run.id == ^run_id and run.state == "queued",
|
|
1223
|
+ |
where: is_nil(run.lease_expires_at) or run.lease_expires_at < ^now,
|
|
1224
|
+ |
select: run
|
|
1225
|
+ |
),
|
|
1226
|
+ |
set: [lease_owner: worker, lease_expires_at: expires_at, updated_at: now],
|
|
1227
|
+ |
inc: [attempt_count: 1]
|
|
1228
|
+ |
)
|
|
1229
|
+ |
end
|
|
1230
|
+ |
|
|
1231
|
+ |
defp approvable(%Run{state: state}) when state in ~w(requested checking waiting_for_approval),
|
|
1232
|
+ |
do: :ok
|
|
1233
|
+ |
|
|
1234
|
+ |
defp approvable(%Run{state: state}), do: {:error, {:illegal_transition, state, "queued"}}
|
|
1235
|
+ |
|
|
1236
|
+ |
# An optimistic precondition: the caller states the state it believes the run
|
|
1237
|
+ |
# holds, and a mismatch is refused rather than applied to a run that moved.
|
|
1238
|
+ |
defp precondition(%Run{} = run, options) do
|
|
1239
|
+ |
case Keyword.get(options, :if_state) do
|
|
1240
|
+ |
nil -> :ok
|
|
1241
|
+ |
state when state == run.state -> :ok
|
|
1242
|
+ |
_mismatch -> {:error, :precondition_failed}
|
|
1243
|
+ |
end
|
|
1244
|
+ |
end
|
|
1245
|
+ |
|
|
1246
|
+ |
defp cancel_requester(%Principal{user: %User{id: id}}), do: id
|
|
1247
|
+ |
defp cancel_requester(%Principal{}), do: nil
|
|
1248
|
+ |
|
|
1249
|
+ |
defp intent(attrs) do
|
|
1250
|
+ |
%{
|
|
1251
|
+ |
source_ref: attrs["source_ref"],
|
|
1252
|
+ |
source_workflow: attrs["source_workflow"],
|
|
1253
|
+ |
workflow_run_id: attrs["workflow_run_id"]
|
|
1254
|
+ |
}
|
|
1255
|
+ |
|> Enum.reject(fn {_key, value} -> is_nil(value) end)
|
|
1256
|
+ |
|> Map.new()
|
|
1257
|
+ |
end
|
|
1258
|
+ |
|
|
1259
|
+ |
defp put_creator(changeset, %Principal{user: %User{id: id}}, %Environment{id: nil}),
|
|
1260
|
+ |
do: Ecto.Changeset.put_change(changeset, :created_by_user_id, id)
|
|
1261
|
+ |
|
|
1262
|
+ |
defp put_creator(changeset, _principal, _environment), do: changeset
|
|
1263
|
+ |
|
|
1264
|
+ |
defp put_publisher(changeset, %Principal{kind: :workflow, grant: %WorkflowGrant{id: id}}),
|
|
1265
|
+ |
do: Ecto.Changeset.put_change(changeset, :published_by_grant_id, id)
|
|
1266
|
+ |
|
|
1267
|
+ |
defp put_publisher(changeset, %Principal{user: %User{id: id}}),
|
|
1268
|
+ |
do: Ecto.Changeset.put_change(changeset, :published_by_user_id, id)
|
|
1269
|
+ |
|
|
1270
|
+ |
defp put_publisher(changeset, %Principal{}), do: changeset
|
|
1271
|
+ |
|
|
1272
|
+ |
defp verify_commit(%Repository{} = repository, commit_sha, commit_store)
|
|
1273
|
+ |
when is_function(commit_store, 2) do
|
|
1274
|
+ |
commit_store.(repository, to_string(commit_sha))
|
|
1275
|
+ |
end
|
|
1276
|
+ |
|
|
1277
|
+ |
# Configurable so a host without git storage — and a test that does not need
|
|
1278
|
+ |
# one — can bind a different store, while the default reads real bytes.
|
|
1279
|
+ |
defp default_commit_store do
|
|
1280
|
+ |
Application.get_env(:openagents, :deployment_commit_store, &commit_exists/2)
|
|
1281
|
+ |
end
|
|
1282
|
+ |
|
|
1283
|
+ |
# The commit must exist in the repository's own storage. Without this, a
|
|
1284
|
+ |
# request can name bytes the repository never received, and every downstream
|
|
1285
|
+ |
# check is then bound to nothing.
|
|
1286
|
+ |
defp commit_exists(%Repository{} = repository, commit_sha) do
|
|
1287
|
+ |
path = OpenAgents.Forge.Repos.bare_path(repository.storage_key)
|
|
1288
|
+ |
|
|
1289
|
+ |
case OpenAgents.Forge.Repos.git(path, ["cat-file", "-e", commit_sha <> "^{commit}"]) do
|
|
1290
|
+ |
{_output, 0} -> :ok
|
|
1291
|
+ |
_missing -> {:error, :unknown_commit}
|
|
1292
|
+ |
end
|
|
1293
|
+ |
end
|
|
1294
|
+ |
|
|
1295
|
+ |
defp grant_environment_id(_repository, nil), do: {:ok, nil}
|
|
1296
|
+ |
|
|
1297
|
+ |
defp grant_environment_id(repository, name) do
|
|
1298
|
+ |
with {:ok, environment} <- fetch_environment_record(repository, name) do
|
|
1299
|
+ |
{:ok, environment.id}
|
|
1300
|
+ |
end
|
|
1301
|
+ |
end
|
|
1302
|
+ |
|
|
1303
|
+ |
defp cursor_run(_repository, nil), do: nil
|
|
1304
|
+ |
|
|
1305
|
+ |
defp cursor_run(%Repository{} = repository, cursor) do
|
|
1306
|
+ |
case fetch_run_record(cursor) do
|
|
1307
|
+ |
{:ok, %Run{repository_id: repository_id} = run} when repository_id == repository.id -> run
|
|
1308
|
+ |
_other -> nil
|
|
1309
|
+ |
end
|
|
1310
|
+ |
end
|
|
1311
|
+ |
|
|
1312
|
+ |
# A grant's lifetime is clamped rather than trusted: a long-lived workflow
|
|
1313
|
+ |
# credential is the thing this design is trying not to have.
|
|
1314
|
+ |
defp grant_lifetime(seconds) when is_integer(seconds) and seconds > 0,
|
|
1315
|
+ |
do: min(seconds, WorkflowGrant.maximum_lifetime_seconds())
|
|
1316
|
+ |
|
|
1317
|
+ |
defp grant_lifetime(_invalid), do: 900
|
|
1318
|
+ |
|
|
1319
|
+ |
defp limit(options) do
|
|
1320
|
+ |
options
|
|
1321
|
+ |
|> Keyword.get(:limit, @default_limit)
|
|
1322
|
+ |
|> case do
|
|
1323
|
+ |
value when is_integer(value) and value > 0 -> min(value, @maximum_limit)
|
|
1324
|
+ |
_invalid -> @default_limit
|
|
1325
|
+ |
end
|
|
1326
|
+ |
end
|
|
1327
|
+ |
|
|
1328
|
+ |
defp bounded_reason(reason) when is_atom(reason),
|
|
1329
|
+ |
do: reason |> Atom.to_string() |> String.slice(0, 80)
|
|
1330
|
+ |
|
|
1331
|
+ |
defp bounded_reason(reason) when is_binary(reason), do: String.slice(reason, 0, 80)
|
|
1332
|
+ |
defp bounded_reason(_reason), do: "provider_failed"
|
|
1333
|
+ |
|
|
1334
|
+ |
defp maybe_put(map, _key, nil), do: map
|
|
1335
|
+ |
defp maybe_put(map, key, value), do: Map.put(map, key, value)
|
|
1336
|
+ |
|
|
1337
|
+ |
defp denil(:nil_value), do: nil
|
|
1338
|
+ |
defp denil(value), do: value
|
|
1339
|
+ |
|
|
1340
|
+ |
defp normalize(attrs) when is_map(attrs) do
|
|
1341
|
+ |
Map.new(attrs, fn {key, value} -> {to_string(key), value} end)
|
|
1342
|
+ |
end
|
|
1343
|
+ |
|
|
1344
|
+ |
defp digest(parts) when is_list(parts) do
|
|
1345
|
+ |
parts
|
|
1346
|
+ |
|> Enum.map_join("\n", &to_string/1)
|
|
1347
|
+ |
|> digest()
|
|
1348
|
+ |
end
|
|
1349
|
+ |
|
|
1350
|
+ |
defp digest(value) when is_binary(value),
|
|
1351
|
+ |
do: :crypto.hash(:sha256, value) |> Base.encode16(case: :lower)
|
|
1352
|
+ |
|
|
1353
|
+ |
@doc "The largest page any listing returns."
|
|
1354
|
+ |
@spec maximum_limit() :: pos_integer()
|
|
1355
|
+ |
def maximum_limit, do: @maximum_limit
|
|
1356
|
+ |
|
|
1357
|
+ |
@doc "The page size a listing uses when the caller does not choose one."
|
|
1358
|
+ |
@spec default_limit() :: pos_integer()
|
|
1359
|
+ |
def default_limit, do: @default_limit
|
|
1360
|
+ |
end
|