|
| 1 | +# SPDX-License-Identifier: Apache-2.0 |
| 2 | + |
| 3 | +import time |
| 4 | + |
| 5 | +from http import HTTPStatus |
| 6 | + |
| 7 | +import pytest |
| 8 | +import responses |
| 9 | + |
| 10 | +from tests.common.db.accounts import UserFactory |
| 11 | +from tests.common.db.organizations import OrganizationFactory, OrganizationRoleFactory |
| 12 | +from warehouse.organizations.models import OrganizationRoleType |
| 13 | +from warehouse.utils.otp import _get_totp |
| 14 | + |
| 15 | + |
| 16 | +@pytest.mark.usefixtures("_enable_all_oidc_providers") |
| 17 | +class TestManageOrganizationPublishing: |
| 18 | + @responses.activate |
| 19 | + def test_add_pending_github_publisher_to_organization(self, webtest): |
| 20 | + """ |
| 21 | + An authenticated user who is an organization owner can add a pending |
| 22 | + GitHub trusted publisher to their organization. |
| 23 | + """ |
| 24 | + # Arrange: Create a user with an organization |
| 25 | + user = UserFactory.create( |
| 26 | + with_verified_primary_email=True, |
| 27 | + with_terms_of_service_agreement=True, |
| 28 | + clear_pwd="password", |
| 29 | + ) |
| 30 | + organization = OrganizationFactory.create(name="test-organization") |
| 31 | + OrganizationRoleFactory.create( |
| 32 | + user=user, |
| 33 | + organization=organization, |
| 34 | + role_name=OrganizationRoleType.Owner, |
| 35 | + ) |
| 36 | + |
| 37 | + # Mock GitHub API for owner validation |
| 38 | + responses.add( |
| 39 | + responses.GET, |
| 40 | + "https://api.github.com/users/test-owner", |
| 41 | + json={ |
| 42 | + "id": 123456, |
| 43 | + "login": "test-owner", |
| 44 | + }, |
| 45 | + status=200, |
| 46 | + ) |
| 47 | + |
| 48 | + # Act: Log in |
| 49 | + login_page = webtest.get("/account/login/", status=HTTPStatus.OK) |
| 50 | + login_form = login_page.forms["login-form"] |
| 51 | + csrf_token = login_form["csrf_token"].value |
| 52 | + login_form["username"] = user.username |
| 53 | + login_form["password"] = "password" |
| 54 | + |
| 55 | + # Handle 2FA |
| 56 | + two_factor_page = login_form.submit().follow(status=HTTPStatus.OK) |
| 57 | + two_factor_form = two_factor_page.forms["totp-auth-form"] |
| 58 | + two_factor_form["csrf_token"] = csrf_token |
| 59 | + two_factor_form["totp_value"] = ( |
| 60 | + _get_totp(user.totp_secret).generate(time.time()).decode() |
| 61 | + ) |
| 62 | + two_factor_form.submit().follow(status=HTTPStatus.OK) |
| 63 | + |
| 64 | + # Navigate to organization publishing page |
| 65 | + publishing_page = webtest.get( |
| 66 | + f"/manage/organization/{organization.normalized_name}/publishing/", |
| 67 | + status=HTTPStatus.OK, |
| 68 | + ) |
| 69 | + |
| 70 | + # Get logged-in CSRF token |
| 71 | + logged_in_csrf_token = publishing_page.html.find( |
| 72 | + "input", {"name": "csrf_token"} |
| 73 | + )["value"] |
| 74 | + |
| 75 | + # Fill out the GitHub pending publisher form |
| 76 | + github_form = publishing_page.forms["pending-github-publisher-form"] |
| 77 | + github_form["csrf_token"] = logged_in_csrf_token |
| 78 | + github_form["project_name"] = "test-org-project" |
| 79 | + github_form["owner"] = "test-owner" |
| 80 | + github_form["repository"] = "test-repo" |
| 81 | + github_form["workflow_filename"] = "release.yml" |
| 82 | + github_form["environment"] = "" # Optional field |
| 83 | + |
| 84 | + # Submit the form, redirects back to the same page on success |
| 85 | + response = github_form.submit(status=HTTPStatus.SEE_OTHER) |
| 86 | + response.follow(status=HTTPStatus.OK) |
| 87 | + |
| 88 | + # Assert: Verify success |
| 89 | + # Check flash messages via the JavaScript endpoint |
| 90 | + flash_messages = webtest.get( |
| 91 | + "/_includes/unauthed/flash-messages/", status=HTTPStatus.OK |
| 92 | + ) |
| 93 | + success_message = flash_messages.html.find( |
| 94 | + "span", {"class": "notification-bar__message"} |
| 95 | + ) |
| 96 | + assert success_message is not None |
| 97 | + assert "Registered a new pending publisher" in success_message.text |
| 98 | + assert "test-org-project" in success_message.text |
| 99 | + assert organization.name in success_message.text |
| 100 | + |
| 101 | + def test_add_pending_gitlab_publisher_to_organization(self, webtest): |
| 102 | + """ |
| 103 | + An authenticated user who is an organization owner can add a pending |
| 104 | + GitLab trusted publisher to their organization. |
| 105 | + """ |
| 106 | + # Arrange: Create a user with an organization |
| 107 | + user = UserFactory.create( |
| 108 | + with_verified_primary_email=True, |
| 109 | + with_terms_of_service_agreement=True, |
| 110 | + clear_pwd="password", |
| 111 | + ) |
| 112 | + organization = OrganizationFactory.create(name="test-organization") |
| 113 | + OrganizationRoleFactory.create( |
| 114 | + user=user, |
| 115 | + organization=organization, |
| 116 | + role_name=OrganizationRoleType.Owner, |
| 117 | + ) |
| 118 | + |
| 119 | + # Act: Log in |
| 120 | + login_page = webtest.get("/account/login/", status=HTTPStatus.OK) |
| 121 | + login_form = login_page.forms["login-form"] |
| 122 | + csrf_token = login_form["csrf_token"].value |
| 123 | + login_form["username"] = user.username |
| 124 | + login_form["password"] = "password" |
| 125 | + |
| 126 | + # Handle 2FA |
| 127 | + two_factor_page = login_form.submit().follow(status=HTTPStatus.OK) |
| 128 | + two_factor_form = two_factor_page.forms["totp-auth-form"] |
| 129 | + two_factor_form["csrf_token"] = csrf_token |
| 130 | + two_factor_form["totp_value"] = ( |
| 131 | + _get_totp(user.totp_secret).generate(time.time()).decode() |
| 132 | + ) |
| 133 | + two_factor_form.submit().follow(status=HTTPStatus.OK) |
| 134 | + |
| 135 | + # Navigate to organization publishing page |
| 136 | + publishing_page = webtest.get( |
| 137 | + f"/manage/organization/{organization.normalized_name}/publishing/", |
| 138 | + status=HTTPStatus.OK, |
| 139 | + ) |
| 140 | + |
| 141 | + # Get logged-in CSRF token |
| 142 | + logged_in_csrf_token = publishing_page.html.find( |
| 143 | + "input", {"name": "csrf_token"} |
| 144 | + )["value"] |
| 145 | + |
| 146 | + # Fill out the GitLab pending publisher form |
| 147 | + gitlab_form = publishing_page.forms["pending-gitlab-publisher-form"] |
| 148 | + gitlab_form["csrf_token"] = logged_in_csrf_token |
| 149 | + gitlab_form["project_name"] = "test-org-gitlab-project" |
| 150 | + gitlab_form["namespace"] = "test-namespace" |
| 151 | + gitlab_form["project"] = "test-project" |
| 152 | + gitlab_form["workflow_filepath"] = ".gitlab-ci.yml" |
| 153 | + gitlab_form["environment"] = "" # Optional field |
| 154 | + # issuer_url is a hidden field with default value |
| 155 | + |
| 156 | + # Submit the form, redirects back to the same page on success |
| 157 | + response = gitlab_form.submit(status=HTTPStatus.SEE_OTHER) |
| 158 | + response.follow(status=HTTPStatus.OK) |
| 159 | + |
| 160 | + # Assert: Verify success |
| 161 | + flash_messages = webtest.get( |
| 162 | + "/_includes/unauthed/flash-messages/", status=HTTPStatus.OK |
| 163 | + ) |
| 164 | + success_message = flash_messages.html.find( |
| 165 | + "span", {"class": "notification-bar__message"} |
| 166 | + ) |
| 167 | + assert success_message is not None |
| 168 | + assert "Registered a new pending publisher" in success_message.text |
| 169 | + assert "test-org-gitlab-project" in success_message.text |
| 170 | + assert organization.name in success_message.text |
| 171 | + |
| 172 | + def test_add_pending_google_publisher_to_organization(self, webtest): |
| 173 | + """ |
| 174 | + An authenticated user who is an organization owner can add a pending |
| 175 | + Google trusted publisher to their organization. |
| 176 | + """ |
| 177 | + # Arrange: Create a user with an organization |
| 178 | + user = UserFactory.create( |
| 179 | + with_verified_primary_email=True, |
| 180 | + with_terms_of_service_agreement=True, |
| 181 | + clear_pwd="password", |
| 182 | + ) |
| 183 | + organization = OrganizationFactory.create(name="test-organization") |
| 184 | + OrganizationRoleFactory.create( |
| 185 | + user=user, |
| 186 | + organization=organization, |
| 187 | + role_name=OrganizationRoleType.Owner, |
| 188 | + ) |
| 189 | + |
| 190 | + # Act: Log in |
| 191 | + login_page = webtest.get("/account/login/", status=HTTPStatus.OK) |
| 192 | + login_form = login_page.forms["login-form"] |
| 193 | + csrf_token = login_form["csrf_token"].value |
| 194 | + login_form["username"] = user.username |
| 195 | + login_form["password"] = "password" |
| 196 | + |
| 197 | + # Handle 2FA |
| 198 | + two_factor_page = login_form.submit().follow(status=HTTPStatus.OK) |
| 199 | + two_factor_form = two_factor_page.forms["totp-auth-form"] |
| 200 | + two_factor_form["csrf_token"] = csrf_token |
| 201 | + two_factor_form["totp_value"] = ( |
| 202 | + _get_totp(user.totp_secret).generate(time.time()).decode() |
| 203 | + ) |
| 204 | + two_factor_form.submit().follow(status=HTTPStatus.OK) |
| 205 | + |
| 206 | + # Navigate to organization publishing page |
| 207 | + publishing_page = webtest.get( |
| 208 | + f"/manage/organization/{organization.normalized_name}/publishing/", |
| 209 | + status=HTTPStatus.OK, |
| 210 | + ) |
| 211 | + |
| 212 | + # Get logged-in CSRF token |
| 213 | + logged_in_csrf_token = publishing_page.html.find( |
| 214 | + "input", {"name": "csrf_token"} |
| 215 | + )["value"] |
| 216 | + |
| 217 | + # Fill out the Google pending publisher form |
| 218 | + google_form = publishing_page.forms["pending-google-publisher-form"] |
| 219 | + google_form["csrf_token"] = logged_in_csrf_token |
| 220 | + google_form["project_name"] = "test-org-google-project" |
| 221 | + google_form["email"] = "test@example.com" |
| 222 | + google_form["sub"] = "" # Optional field |
| 223 | + |
| 224 | + # Submit the form, redirects back to the same page on success |
| 225 | + response = google_form.submit(status=HTTPStatus.SEE_OTHER) |
| 226 | + response.follow(status=HTTPStatus.OK) |
| 227 | + |
| 228 | + # Assert: Verify success |
| 229 | + flash_messages = webtest.get( |
| 230 | + "/_includes/unauthed/flash-messages/", status=HTTPStatus.OK |
| 231 | + ) |
| 232 | + success_message = flash_messages.html.find( |
| 233 | + "span", {"class": "notification-bar__message"} |
| 234 | + ) |
| 235 | + assert success_message is not None |
| 236 | + assert "Registered a new pending publisher" in success_message.text |
| 237 | + assert "test-org-google-project" in success_message.text |
| 238 | + assert organization.name in success_message.text |
| 239 | + |
| 240 | + @responses.activate |
| 241 | + def test_add_pending_activestate_publisher_to_organization(self, webtest): |
| 242 | + """ |
| 243 | + An authenticated user who is an organization owner can add a pending |
| 244 | + ActiveState trusted publisher to their organization. |
| 245 | + """ |
| 246 | + # Arrange: Create a user with an organization |
| 247 | + user = UserFactory.create( |
| 248 | + with_verified_primary_email=True, |
| 249 | + with_terms_of_service_agreement=True, |
| 250 | + clear_pwd="password", |
| 251 | + ) |
| 252 | + organization = OrganizationFactory.create(name="test-organization") |
| 253 | + OrganizationRoleFactory.create( |
| 254 | + user=user, |
| 255 | + organization=organization, |
| 256 | + role_name=OrganizationRoleType.Owner, |
| 257 | + ) |
| 258 | + |
| 259 | + # Mock ActiveState API for organization and actor validation |
| 260 | + # The form makes two sequential API calls: |
| 261 | + # 1. Organization validation (validate_organization method) |
| 262 | + # 2. Actor validation (validate_actor method) |
| 263 | + responses.add( |
| 264 | + responses.POST, |
| 265 | + "https://platform.activestate.com/graphql/v1/graphql", |
| 266 | + json={"data": {"organizations": [{"added": "2020-01-01"}]}}, |
| 267 | + status=200, |
| 268 | + ) |
| 269 | + responses.add( |
| 270 | + responses.POST, |
| 271 | + "https://platform.activestate.com/graphql/v1/graphql", |
| 272 | + json={"data": {"users": [{"user_id": "test-user-id"}]}}, |
| 273 | + status=200, |
| 274 | + ) |
| 275 | + |
| 276 | + # Act: Log in |
| 277 | + login_page = webtest.get("/account/login/", status=HTTPStatus.OK) |
| 278 | + login_form = login_page.forms["login-form"] |
| 279 | + csrf_token = login_form["csrf_token"].value |
| 280 | + login_form["username"] = user.username |
| 281 | + login_form["password"] = "password" |
| 282 | + |
| 283 | + # Handle 2FA |
| 284 | + two_factor_page = login_form.submit().follow(status=HTTPStatus.OK) |
| 285 | + two_factor_form = two_factor_page.forms["totp-auth-form"] |
| 286 | + two_factor_form["csrf_token"] = csrf_token |
| 287 | + two_factor_form["totp_value"] = ( |
| 288 | + _get_totp(user.totp_secret).generate(time.time()).decode() |
| 289 | + ) |
| 290 | + two_factor_form.submit().follow(status=HTTPStatus.OK) |
| 291 | + |
| 292 | + # Navigate to organization publishing page |
| 293 | + publishing_page = webtest.get( |
| 294 | + f"/manage/organization/{organization.normalized_name}/publishing/", |
| 295 | + status=HTTPStatus.OK, |
| 296 | + ) |
| 297 | + |
| 298 | + # Get logged-in CSRF token |
| 299 | + logged_in_csrf_token = publishing_page.html.find( |
| 300 | + "input", {"name": "csrf_token"} |
| 301 | + )["value"] |
| 302 | + |
| 303 | + # Fill out the ActiveState pending publisher form |
| 304 | + activestate_form = publishing_page.forms["pending-activestate-publisher-form"] |
| 305 | + activestate_form["csrf_token"] = logged_in_csrf_token |
| 306 | + activestate_form["project_name"] = "test-org-activestate-project" |
| 307 | + activestate_form["organization"] = "test-activestate-org" |
| 308 | + activestate_form["project"] = "test-activestate-project" |
| 309 | + activestate_form["actor"] = "test-actor" |
| 310 | + |
| 311 | + # Submit the form, redirects back to the same page on success |
| 312 | + response = activestate_form.submit(status=HTTPStatus.SEE_OTHER) |
| 313 | + response.follow(status=HTTPStatus.OK) |
| 314 | + |
| 315 | + # Assert: Verify success |
| 316 | + flash_messages = webtest.get( |
| 317 | + "/_includes/unauthed/flash-messages/", status=HTTPStatus.OK |
| 318 | + ) |
| 319 | + success_message = flash_messages.html.find( |
| 320 | + "span", {"class": "notification-bar__message"} |
| 321 | + ) |
| 322 | + assert success_message is not None |
| 323 | + assert "Registered a new pending publisher" in success_message.text |
| 324 | + assert "test-org-activestate-project" in success_message.text |
| 325 | + assert organization.name in success_message.text |
0 commit comments