Settings → Runners for repository admins and organisation owners (refs #41, part 2b) 7974adb0be66

/// The runners that may take a repository's jobs: the site's, its owner's

/// and its own (oxygenhub#41).

pub async fn serving(pool: &Pool, owner: &str, repo_id: i64) -> sqlx::Result<Vec<Runner>> {

    sqlx::query_as(

        "SELECT * FROM flow_runners WHERE (scope_owner = '' AND scope_repo_id IS NULL) \

         OR (scope_owner = ? AND scope_repo_id IS NULL) OR scope_repo_id = ? ORDER BY name",

    )

    .bind(owner)

    .bind(repo_id)

    .fetch_all(pool)

    .await

}

/// The runners that serve an owner: the site's, and those scoped to the

/// owner or one of its repositories.

pub async fn serving_owner(pool: &Pool, owner: &str) -> sqlx::Result<Vec<Runner>> {

    sqlx::query_as("SELECT * FROM flow_runners WHERE (scope_owner = '' AND scope_repo_id IS NULL) OR scope_owner = ? ORDER BY name")

        .bind(owner)

        .fetch_all(pool)

        .await

}

        .route("/orgs/{name}/settings/runners/token", post(orgs::runner_token))

        .route("/orgs/{name}/settings/runners/{id}/remove", post(orgs::runner_remove))

        .route("/{owner}/{name}/settings/runners/token", post(repo_settings::runner_token))

        .route("/{owner}/{name}/settings/runners/{id}/remove", post(repo_settings::runner_remove))

    /// Runners serving the organisation: (runner, what it serves, whether

    /// it is the organisation's own).

    runners: Vec<(crate::db::runners::Runner, String, bool)>,

    /// A registration token just made, shown once.

    runner_token: Option<String>,

    let mut runners = Vec::new();

    for r in crate::db::runners::serving_owner(&app.db, &org.name).await.map_err(|e| Error::from_db(shell.clone(), e))? {

        let own = r.scope_owner == org.name;

        let serves = match r.scope_repo_id {

            Some(id) => crate::db::repos::get_by_id(&app.db, id).await.ok().flatten().map(|p| p.slug()).unwrap_or_default(),

            None if own => "the organisation's repositories".into(),

            None => "the whole site".into(),

        };

        runners.push((r, serves, own));

    }

    Ok(Page(OrgSettingsPage { shell, org, members, teams, hooks, notice, error, runners, runner_token: None }))

}

/// `POST /orgs/{name}/settings/runners/token`: a one-time registration

/// token for a runner that serves the organisation's repositories

/// (oxygenhub#41).

pub async fn runner_token(

    State(app): State<Shared>,

    viewer: Viewer,

    Path(name): Path<String>,

    Form(f): Form<super::session::CsrfForm>,

) -> Result<Page<OrgSettingsPage>, Error> {

    let shell = Shell::new(&app, &format!("/orgs/{name}/settings"), &viewer);

    let (me, org) = owned_org(&app, &shell, &viewer, &name).await?;

    csrf(&viewer, &f.csrf, &shell)?;

    let token = format!("flowreg_{}", crate::auth::random_hex(24));

    let scope = crate::db::runners::Scope { owner: org.name.clone(), repo_id: None };

    crate::db::runners::add_registration(&app.db, &blake3::hash(token.as_bytes()).to_hex(), &me.username, &scope)

        .await

        .map_err(|e| Error::from_db(shell.clone(), e))?;

    accounts::audit(&app.db, Some(&me), "flow.runner.token", &org.name, "registration token for the organisation").await.ok();

    let mut page = org_page(&app, shell, org, Some("Copy the token now; it is not shown again.".into()), None).await?;

    page.0.runner_token = Some(token);

    Ok(page)

}

/// `POST /orgs/{name}/settings/runners/{id}/remove`: remove a runner scoped

/// to the organisation or one of its repositories, when it runs nothing.

pub async fn runner_remove(

    State(app): State<Shared>,

    viewer: Viewer,

    Path((name, id)): Path<(String, i64)>,

    Form(f): Form<super::session::CsrfForm>,

) -> Result<Page<OrgSettingsPage>, Error> {

    let shell = Shell::new(&app, &format!("/orgs/{name}/settings"), &viewer);

    let (me, org) = owned_org(&app, &shell, &viewer, &name).await?;

    csrf(&viewer, &f.csrf, &shell)?;

    let e = |e| Error::from_db(shell.clone(), e);

    let Some(r) = crate::db::runners::get(&app.db, id).await.map_err(e)?.filter(|r| r.scope_owner == org.name) else {

        return Err(Error::not_found(shell));

    };

    if crate::db::runners::running(&app.db).await.map_err(e)?.iter().any(|(rid, n)| *rid == id && *n > 0) {

        return org_page(&app, shell, org, None, Some(format!("{} is running jobs: wait for them to finish.", r.name))).await;

    }

    crate::db::runners::remove(&app.db, id).await.map_err(e)?;

    accounts::audit(&app.db, Some(&me), "flow.runner.removed", &r.name, &org.name).await.ok();

    org_page(&app, shell, org, Some(format!("{} removed: its credential no longer works.", r.name)), None).await

    /// Runners that may take this repository's jobs: (runner, what it

    /// serves, whether it is this repository's own).

    runners: Vec<(db::runners::Runner, String, bool)>,

    /// A registration token just made, shown once.

    runner_token: Option<String>,

    let runners = db::runners::serving(&app.db, &row.owner, row.id)

        .await

        .map_err(|e| Error::from_db(shell.clone(), e))?

        .into_iter()

        .map(|r| {

            let own = r.scope_repo_id == Some(row.id);

            let serves = if own {

                "this repository".to_string()

            } else if r.scope_owner.is_empty() {

                "the whole site".to_string()

            } else {

                format!("{}'s repositories", r.scope_owner)

            };

            (r, serves, own)

        })

        .collect();

        runners,

        runner_token: None,

}

/// `POST /{owner}/{name}/settings/runners/token`: a one-time registration

/// token for a runner that serves this repository only (oxygenhub#41).

pub async fn runner_token(

    State(app): State<Shared>,

    viewer: Viewer,

    Path((owner, name)): Path<(String, String)>,

    Form(f): Form<super::session::CsrfForm>,

) -> Result<Page<RepoSettingsPage>, Error> {

    let shell = Shell::new(&app, &format!("/{owner}/{name}/settings"), &viewer);

    let (_, row, role) = lookup_role(&app, &shell, &viewer, &owner, &name, Access::Admin).await?;

    csrf(&viewer, &f.csrf, &shell)?;

    let me = viewer.account().map(|a| a.username.clone()).unwrap_or_default();

    let token = format!("flowreg_{}", crate::auth::random_hex(24));

    let scope = db::runners::Scope { owner: row.owner.clone(), repo_id: Some(row.id) };

    db::runners::add_registration(&app.db, &blake3::hash(token.as_bytes()).to_hex(), &me, &scope)

        .await

        .map_err(|e| Error::from_db(shell.clone(), e))?;

    accounts::audit(&app.db, viewer.account(), "flow.runner.token", &row.slug(), "registration token for this repository").await.ok();

    let mut page = settings_page(&app, shell, row, role, Some("Copy the token now; it is not shown again.".into()), None).await?;

    page.0.runner_token = Some(token);

    Ok(page)

}

/// `POST /{owner}/{name}/settings/runners/{id}/remove`: remove one of this

/// repository's own runners, when it is running nothing.

pub async fn runner_remove(

    State(app): State<Shared>,

    viewer: Viewer,

    Path((owner, name, id)): Path<(String, String, i64)>,

    Form(f): Form<super::session::CsrfForm>,

) -> Result<Page<RepoSettingsPage>, Error> {

    let shell = Shell::new(&app, &format!("/{owner}/{name}/settings"), &viewer);

    let (_, row, role) = lookup_role(&app, &shell, &viewer, &owner, &name, Access::Admin).await?;

    csrf(&viewer, &f.csrf, &shell)?;

    let e = |e| Error::from_db(shell.clone(), e);

    let Some(r) = db::runners::get(&app.db, id).await.map_err(e)?.filter(|r| r.scope_repo_id == Some(row.id)) else {

        return Err(Error::not_found(shell));

    };

    if db::runners::running(&app.db).await.map_err(e)?.iter().any(|(rid, n)| *rid == id && *n > 0) {

        return settings_page(&app, shell, row, role, None, Some(format!("{} is running jobs: wait for them to finish.", r.name))).await;

    }

    db::runners::remove(&app.db, id).await.map_err(e)?;

    accounts::audit(&app.db, viewer.account(), "flow.runner.removed", &r.name, &row.slug()).await.ok();

    settings_page(&app, shell, row, role, Some(format!("{} removed: its credential no longer works.", r.name)), None).await

/// Settings → Runners (oxygenhub#41, part 2b): a repository's admin adds a

/// runner for that repository alone, sees which runners serve it, and

/// removes only its own.

#[tokio::test(flavor = "multi_thread")]

async fn repository_admins_add_and_remove_their_own_runners() {

    use oxy_core::identity::SecretKey;

    let h = hub().await;

    let res = oxygenhub::router(h.app.clone())

        .oneshot(

            Request::post("/register")

                .header(header::CONTENT_TYPE, "application/x-www-form-urlencoded")

                .body(Body::from("username=andy&email=andy%40example.com&password=correct-horse-battery"))

                .unwrap(),

        )

        .await

        .unwrap();

    let cookie = res.headers().get(header::SET_COOKIE).unwrap().to_str().unwrap().split(';').next().unwrap().to_string();

    create(&h.app, "andy", "demo").await;

    let demo = db::repos::get(&h.app.db, "andy", "demo").await.unwrap().unwrap();

    let web = |method: &'static str, path: String, body: Option<String>| {

        let (app, cookie) = (h.app.clone(), cookie.clone());

        async move {

            let mut req = Request::builder().method(method).uri(path).header(header::COOKIE, cookie);

            if body.is_some() {

                req = req.header(header::CONTENT_TYPE, "application/x-www-form-urlencoded");

            }

            let res = oxygenhub::router(app).oneshot(req.body(body.map(Body::from).unwrap_or_else(Body::empty)).unwrap()).await.unwrap();

            let st = res.status();

            (st, String::from_utf8_lossy(&res.into_body().collect().await.unwrap().to_bytes()).into_owned())

        }

    };

    let csrf_of = |page: &str| {

        let i = page.find("name=\"_csrf\" value=\"").unwrap() + "name=\"_csrf\" value=\"".len();

        page[i..].split('"').next().unwrap().to_string()

    };

    let (st, page) = web("GET", "/andy/demo/settings".into(), None).await;

    assert_eq!(st, StatusCode::OK);

    assert!(page.contains("<h2>Runners</h2>"), "the owner sees the section");

    let (_, page) = web("POST", "/andy/demo/settings/runners/token".into(), Some(format!("_csrf={}", csrf_of(&page)))).await;

    let i = page.find("--token ").unwrap() + "--token ".len();

    let token: String = page[i..].chars().take_while(|c| !c.is_whitespace()).collect();

    let identity = SecretKey::generate().identity("mine", "", 1);

    let info = serde_json::json!({ "arch": "x86_64", "os": "linux", "version": "0.1.0", "labels": [], "max_jobs": 1 });

    let reg = serde_json::json!({ "token": token, "name": "mine", "identity": hex::encode(identity.encode()), "info": info });

    let (st, body) = call(&h.app, "POST", "/api/v1/flow/runners/register", None, Some(reg.to_string())).await;

    assert_eq!(st, StatusCode::OK, "{body}");

    let mine = db::runners::list(&h.app.db).await.unwrap().into_iter().find(|r| r.name == "mine").unwrap();

    assert_eq!(mine.scope(), db::runners::Scope { owner: "andy".into(), repo_id: Some(demo.id) });

    // The site's runner (hub.toml, from its first claim) serves it too, but is not the owner's to remove.

    claim(&h.app).await;

    let site = db::runners::list(&h.app.db).await.unwrap().into_iter().find(|r| r.name == "test-1").unwrap();

    let (_, page) = web("GET", "/andy/demo/settings".into(), None).await;

    assert!(page.contains("<strong>mine</strong>") && page.contains("this repository"), "{page}");

    assert!(page.contains("<strong>test-1</strong>") && page.contains("the whole site"), "{page}");

    let csrf = csrf_of(&page);

    let (st, _) = web("POST", format!("/andy/demo/settings/runners/{}/remove", site.id), Some(format!("_csrf={csrf}"))).await;

    assert_eq!(st, StatusCode::NOT_FOUND, "not the repository's own");

    let (_, page) = web("POST", format!("/andy/demo/settings/runners/{}/remove", mine.id), Some(format!("_csrf={csrf}"))).await;

    assert!(page.contains("mine removed"), "{page}");

    assert!(db::runners::get(&h.app.db, mine.id).await.unwrap().is_none());

    // The same for an organisation's owner.

    let (_, page) = web("GET", "/orgs/new".into(), None).await;

    web("POST", "/orgs/new".into(), Some(format!("_csrf={}&name=acme&display_name=Acme", csrf_of(&page)))).await;

    let (st, page) = web("GET", "/orgs/acme/settings".into(), None).await;

    assert_eq!(st, StatusCode::OK);

    let (_, page) = web("POST", "/orgs/acme/settings/runners/token".into(), Some(format!("_csrf={}", csrf_of(&page)))).await;

    let i = page.find("--token ").unwrap() + "--token ".len();

    let token: String = page[i..].chars().take_while(|c| !c.is_whitespace()).collect();

    let identity = SecretKey::generate().identity("acme-runner", "", 1);

    let reg = serde_json::json!({ "token": token, "name": "acme-runner", "identity": hex::encode(identity.encode()), "info": info });

    assert_eq!(call(&h.app, "POST", "/api/v1/flow/runners/register", None, Some(reg.to_string())).await.0, StatusCode::OK);

    let acme = db::runners::list(&h.app.db).await.unwrap().into_iter().find(|r| r.name == "acme-runner").unwrap();

    assert_eq!(acme.scope(), db::runners::Scope { owner: "acme".into(), repo_id: None });

    let (_, page) = web("GET", "/orgs/acme/settings".into(), None).await;

    assert!(page.contains("<strong>acme-runner</strong>") && page.contains("the organisation's repositories"), "{page}");

    let (_, page) = web("POST", format!("/orgs/acme/settings/runners/{}/remove", acme.id), Some(format!("_csrf={}", csrf_of(&page)))).await;

    assert!(page.contains("acme-runner removed"), "{page}");

}

  - **Scopes** (oxygenhub#41): a runner serves the whole site, one owner's repositories (a user or an organisation), or one repository. The registration token carries the scope (typed as empty, `owner` or `owner/repository` when it is made) and the runner gets it; an administrator can change it on the Runners page. A claim returns only queued jobs of repositories in the runner's scope, filtered before the queue is cut short, so a scoped runner never receives another scope's jobs and never waits behind them. `hub.toml` runners serve the whole site. Changes are audited (`flow.runner.scope`). **Settings → Runners** on a repository (its admins) and on an organisation (its owners) lists the runners that may take its jobs, with what each serves; there they make a registration token for a runner of their own scope (the repository, or the organisation's repositories) and remove such runners when they run nothing. Only site administrators manage site runners.

<section class="panel" id="runners">

  <h2>Runners</h2>

  <p class="small">The OxygenFlow runners that may take this organisation's jobs: the site's, the organisation's, and those of its repositories.</p>

  {% if let Some(t) = runner_token %}<p class="note">Register a runner for the organisation with:</p><pre class="snippet">flow runner register {{ shell.base_url }} --token {{ t }} --name NAME</pre>{% endif %}

  {% if runners.is_empty() %}<p class="mute small">None yet: runs wait until a runner that serves them is added.</p>{% else %}

  <div class="table"><table>

  <tr><th>Runner</th><th>Status</th><th>Serves</th><th>Machine</th><th></th></tr>

  {% for (r, serves, own) in runners %}<tr><td><strong>{{ r.name }}</strong></td><td><span class="runner-status st-{{ r.status() }}">{{ r.status() }}</span></td><td class="small">{{ serves }}</td><td class="small">{% if r.arch.is_empty() %}<span class="mute">unknown</span>{% else %}{{ r.arch }} · {{ r.os }}{% endif %}</td>

  <td>{% if *own %}<form class="tiny" method="post" action="/orgs/{{ org.name }}/settings/runners/{{ r.id }}/remove" onsubmit="return confirm('Remove {{ r.name }}? Its credential stops working.')"><input type="hidden" name="_csrf" value="{{ shell.csrf }}"><button class="danger-btn">remove</button></form>{% endif %}</td></tr>{% endfor %}

  </table></div>{% endif %}

  <form method="post" action="/orgs/{{ org.name }}/settings/runners/token"><input type="hidden" name="_csrf" value="{{ shell.csrf }}"><button type="submit">Add a runner for the organisation</button></form>

</section>

<section class="panel" id="runners">

  <h2>Runners</h2>

  <p class="small">The OxygenFlow runners that may take this repository's jobs: the site's, {{ row.owner }}'s, and this repository's own.</p>

  {% if let Some(t) = runner_token %}<p class="note">Register a runner for this repository with:</p><pre class="snippet">flow runner register {{ shell.base_url }} --token {{ t }} --name NAME</pre>{% endif %}

  {% if runners.is_empty() %}<p class="mute small">None yet: runs wait until a runner that serves this repository is added.</p>{% else %}

  <div class="table"><table>

  <tr><th>Runner</th><th>Status</th><th>Serves</th><th>Machine</th><th></th></tr>

  {% for (r, serves, own) in runners %}<tr><td><strong>{{ r.name }}</strong></td><td><span class="runner-status st-{{ r.status() }}">{{ r.status() }}</span></td><td class="small">{{ serves }}</td><td class="small">{% if r.arch.is_empty() %}<span class="mute">unknown</span>{% else %}{{ r.arch }} · {{ r.os }}{% endif %}</td>

  <td>{% if *own %}<form class="tiny" method="post" action="{{ repo.url }}/settings/runners/{{ r.id }}/remove" onsubmit="return confirm('Remove {{ r.name }}? Its credential stops working.')"><input type="hidden" name="_csrf" value="{{ shell.csrf }}"><button class="danger-btn">remove</button></form>{% endif %}</td></tr>{% endfor %}

  </table></div>{% endif %}

  <form method="post" action="{{ repo.url }}/settings/runners/token"><input type="hidden" name="_csrf" value="{{ shell.csrf }}"><button type="submit">Add a runner for this repository</button></form>

</section>