Compare commits

..

2 Commits

Author SHA1 Message Date
Anthony LC 2ac11f27fb 🚨(app-desk) add duration on TextErrors component
Add a duration on the TextErrors component.
I will display the error message the time specified
in the `duration` prop.
By default the errors will be displayed 6 seconds.
2024-03-06 11:03:49 +01:00
Anthony LC d983397a34 💫(app-desk) add effect prop on Box component
Add the effect prop on Box component.
2 effects are available: 'show' and 'hide'.
2024-03-06 10:56:54 +01:00
557 changed files with 16694 additions and 48169 deletions
-3
View File
@@ -31,6 +31,3 @@ db.sqlite3
.mypy_cache
.pylint.d
.pytest_cache
# Frontend dependencies
node_modules
-35
View File
@@ -1,35 +0,0 @@
name: Deploy
on:
push:
tags:
- 'preprod'
- 'production'
jobs:
notify-argocd:
runs-on: ubuntu-latest
steps:
-
name: Checkout repository
uses: actions/checkout@v4
-
name: Call argocd github webhook
run: |
data='{"ref": "'$GITHUB_REF'","repository": {"html_url":"'$GITHUB_SERVER_URL'/'$GITHUB_REPOSITORY'"}}'
sig=$(echo -n ${data} | openssl dgst -sha1 -hmac ''${ARGOCD_WEBHOOK_SECRET}'' | awk '{print "X-Hub-Signature: sha1="$2}')
curl -X POST -H 'X-GitHub-Event:push' -H "Content-Type: application/json" -H "${sig}" --data "${data}" $ARGOCD_WEBHOOK_URL
sig=$(echo -n ${data} | openssl dgst -sha1 -hmac ''${ARGOCD_PRODUCTION_WEBHOOK_SECRET}'' | awk '{print "X-Hub-Signature: sha1="$2}')
curl -X POST -H 'X-GitHub-Event:push' -H "Content-Type: application/json" -H "${sig}" --data "${data}" $ARGOCD_PRODUCTION_WEBHOOK_URL
start-test-on-preprod:
needs:
- notify-argocd
runs-on: ubuntu-latest
if: startsWith(github.event.ref, 'refs/tags/preprod')
steps:
-
name: Debug
run: |
echo "Start test when preprod is ready"
+9 -94
View File
@@ -1,5 +1,4 @@
name: Docker Hub Workflow
run-name: Docker Hub Workflow
on:
workflow_dispatch:
@@ -12,118 +11,34 @@ on:
branches:
- 'main'
env:
DOCKER_USER: 1001:127
jobs:
trivy-scan:
build-and-push:
runs-on: ubuntu-latest
steps:
-
name: Checkout repository
name: Checkout
uses: actions/checkout@v4
-
name: Docker meta
id: meta
uses: docker/metadata-action@v5
with:
images: |
lasuite/people-backend
lasuite/people-frontend
images: lasuite/people
-
name: Run trivy scan (backend)
uses: numerique-gouv/action-trivy-cache@main
name: Load sops secrets
uses: rouja/actions-sops@main
with:
docker-build-args: '--target backend-production -f Dockerfile'
docker-image-name: 'docker.io/lasuite/people-backend:${{ github.sha }}'
-
name: Run trivy scan (frontend)
uses: numerique-gouv/action-trivy-cache@main
with:
docker-build-args: '--target frontend-production -f Dockerfile'
docker-image-name: 'docker.io/lasuite/people-frontend:${{ github.sha }}'
build-and-push-backend:
runs-on: ubuntu-latest
steps:
-
name: Checkout repository
uses: actions/checkout@v4
-
name: Docker meta
id: meta
uses: docker/metadata-action@v5
with:
images: lasuite/people-backend
secret-file: .github/workflows/secrets.enc.env
age-key: ${{ secrets.SOPS_PRIVATE }}
-
name: Login to DockerHub
if: github.event_name != 'pull_request'
run: echo "${{ secrets.DOCKER_HUB_PASSWORD }}" | docker login -u "${{ secrets.DOCKER_HUB_USER }}" --password-stdin
- name: create-version-json
id: create-version-json
uses: jsdaniell/create-json@v1.2.3
with:
name: "version.json"
json: '{"source":"${{github.repository}}", "version":"${{github.ref_name}}", "commit":"${{github.sha}}", "build": "NA"}'
dir: 'src/backend'
run: echo "$DOCKER_HUB_PASSWORD" | docker login -u "$DOCKER_HUB_USER" --password-stdin
-
name: Build and push
uses: docker/build-push-action@v6
uses: docker/build-push-action@v5
with:
context: .
target: backend-production
build-args: DOCKER_USER=${{ env.DOCKER_USER }}:-1000
push: ${{ github.event_name != 'pull_request' }}
tags: ${{ steps.meta.outputs.tags }}
labels: ${{ steps.meta.outputs.labels }}
build-and-push-frontend:
runs-on: ubuntu-latest
steps:
-
name: Checkout repository
uses: actions/checkout@v4
-
name: Docker meta
id: meta
uses: docker/metadata-action@v5
with:
images: lasuite/people-frontend
- name: create-version-json
id: create-version-json
uses: jsdaniell/create-json@v1.2.3
with:
name: "version.json"
json: '{"source":"${{github.repository}}", "version":"${{github.ref_name}}", "commit":"${{github.sha}}", "build": "NA"}'
dir: 'src/frontend/apps/desk'
-
name: Login to DockerHub
if: github.event_name != 'pull_request'
run: echo "${{ secrets.DOCKER_HUB_PASSWORD }}" | docker login -u "${{ secrets.DOCKER_HUB_USER }}" --password-stdin
-
name: Build and push
uses: docker/build-push-action@v6
with:
context: .
target: frontend-production
build-args: DOCKER_USER=${{ env.DOCKER_USER }}:-1000
push: ${{ github.event_name != 'pull_request' }}
tags: ${{ steps.meta.outputs.tags }}
labels: ${{ steps.meta.outputs.labels }}
notify-argocd:
needs:
- build-and-push-frontend
- build-and-push-backend
runs-on: ubuntu-latest
if: github.event_name != 'pull_request'
steps:
-
name: Checkout repository
uses: actions/checkout@v4
-
name: Call argocd github webhook
run: |
data='{"ref": "'$GITHUB_REF'","repository": {"html_url":"'$GITHUB_SERVER_URL'/numerique-gouv/lasuite-deploiement"}}'
sig=$(echo -n ${data} | openssl dgst -sha1 -hmac "${{ secrets.ARGOCD_PREPROD_WEBHOOK_SECRET }}" | awk '{print "X-Hub-Signature: sha1="$2}')
curl -X POST -H 'X-GitHub-Event:push' -H "Content-Type: application/json" -H "${sig}" --data "${data}" ${{ vars.ARGOCD_PREPROD_WEBHOOK_URL }}
+116 -142
View File
@@ -4,6 +4,8 @@ on:
push:
branches:
- main
tags:
- 'v*'
pull_request:
branches:
- '*'
@@ -14,9 +16,9 @@ jobs:
if: github.event_name == 'pull_request' # Makes sense only for pull requests
steps:
- name: Checkout repository
uses: actions/checkout@v4
uses: actions/checkout@v2
with:
fetch-depth: 0
fetch-depth: 0
- name: show
run: git log
- name: Enforce absence of print statements in code
@@ -37,9 +39,9 @@ jobs:
github.event_name == 'pull_request'
steps:
- name: Checkout repository
uses: actions/checkout@v4
uses: actions/checkout@v2
with:
fetch-depth: 0
fetch-depth: 0
- name: Check that the CHANGELOG has been modified in the current branch
run: git whatchanged --name-only --pretty="" origin/${{ github.event.pull_request.base.ref }}..HEAD | grep CHANGELOG
@@ -47,7 +49,7 @@ jobs:
runs-on: ubuntu-latest
steps:
- name: Checkout repository
uses: actions/checkout@v4
uses: actions/checkout@v2
- name: Check CHANGELOG max line length
run: |
max_line_length=$(cat CHANGELOG.md | grep -Ev "^\[.*\]: https://github.com" | wc -L)
@@ -56,9 +58,11 @@ jobs:
exit 1
fi
install-front:
test-front:
runs-on: ubuntu-latest
defaults:
run:
working-directory: src/frontend/
steps:
- name: Checkout repository
uses: actions/checkout@v4
@@ -67,173 +71,90 @@ jobs:
uses: actions/setup-node@v4
with:
node-version: '18.x'
- name: Restore the frontend cache
uses: actions/cache@v4
id: front-node_modules
with:
path: 'src/frontend/**/node_modules'
key: front-node_modules-${{ hashFiles('src/frontend/**/yarn.lock') }}
cache: 'yarn'
cache-dependency-path: src/frontend/yarn.lock
- name: Install dependencies
if: steps.front-node_modules.outputs.cache-hit != 'true'
run: cd src/frontend/ && yarn install --frozen-lockfile
- name: Cache install frontend
if: steps.front-node_modules.outputs.cache-hit != 'true'
uses: actions/cache@v4
with:
path: 'src/frontend/**/node_modules'
key: front-node_modules-${{ hashFiles('src/frontend/**/yarn.lock') }}
build-front:
runs-on: ubuntu-latest
needs: install-front
steps:
- name: Checkout repository
uses: actions/checkout@v4
- name: Restore the frontend cache
uses: actions/cache@v4
id: front-node_modules
with:
path: 'src/frontend/**/node_modules'
key: front-node_modules-${{ hashFiles('src/frontend/**/yarn.lock') }}
- name: Build CI App
run: cd src/frontend/ && yarn ci:build
- name: Cache build frontend
uses: actions/cache@v4
with:
path: src/frontend/apps/desk/out/
key: build-front-${{ github.run_id }}
test-front:
runs-on: ubuntu-latest
needs: install-front
steps:
- name: Checkout repository
uses: actions/checkout@v4
- name: Restore the frontend cache
uses: actions/cache@v4
id: front-node_modules
with:
path: 'src/frontend/**/node_modules'
key: front-node_modules-${{ hashFiles('src/frontend/**/yarn.lock') }}
run: yarn install --frozen-lockfile
- name: Test App
run: cd src/frontend/ && yarn app:test
run: yarn app:test
- name: Test Translations
run: yarn i18n:test
lint-front:
runs-on: ubuntu-latest
needs: install-front
defaults:
run:
working-directory: src/frontend/
steps:
- name: Checkout repository
uses: actions/checkout@v4
- name: Restore the frontend cache
uses: actions/cache@v4
id: front-node_modules
- name: Setup Node.js
uses: actions/setup-node@v4
with:
path: 'src/frontend/**/node_modules'
key: front-node_modules-${{ hashFiles('src/frontend/**/yarn.lock') }}
node-version: '18.x'
cache: 'yarn'
cache-dependency-path: src/frontend/yarn.lock
- name: Install dependencies
run: yarn install --frozen-lockfile
- name: Check linting
run: cd src/frontend/ && yarn lint
run: yarn lint
test-e2e:
runs-on: ubuntu-latest
needs: [build-mails, build-front]
timeout-minutes: 10
strategy:
fail-fast: false
matrix:
shardIndex: [1, 2, 3, 4]
shardTotal: [4]
steps:
- name: Checkout repository
uses: actions/checkout@v4
- name: Set services env variables
run: |
make create-env-files
cat env.d/development/common.e2e.dist >> env.d/development/common
- name: Download mails' templates
uses: actions/download-artifact@v4
with:
name: mails-templates
path: src/backend/core/templates/mail
- name: Restore the frontend cache
uses: actions/cache@v4
id: front-node_modules
with:
path: 'src/frontend/**/node_modules'
key: front-node_modules-${{ hashFiles('src/frontend/**/yarn.lock') }}
- name: Restore the build cache
uses: actions/cache@v4
id: cache-build
with:
path: src/frontend/apps/desk/out/
key: build-front-${{ github.run_id }}
- name: Build and Start Docker Servers
env:
DOCKER_BUILDKIT: 1
COMPOSE_DOCKER_CLI_BUILD: 1
run: |
docker compose build --pull --build-arg BUILDKIT_INLINE_CACHE=1
docker-compose build --pull --build-arg BUILDKIT_INLINE_CACHE=1
make run
- name: Apply DRF migrations
run: |
make migrate
- name: Add dummy data
run: |
make demo FLUSH_ARGS='--no-input'
- name: Setup Node.js
uses: actions/setup-node@v4
with:
node-version: '18.x'
cache: 'yarn'
cache-dependency-path: src/frontend/yarn.lock
- name: Setup Dimail DB
run: |
make dimail-setup-db
- name: Install dependencies
run: cd src/frontend/ && yarn install --frozen-lockfile
- name: Install Playwright Browsers
run: cd src/frontend/apps/e2e && yarn install
- name: Build CI App
run: cd src/frontend/ && yarn ci:build
- name: Run e2e tests
run: cd src/frontend/ && yarn e2e:test --shard=${{ matrix.shardIndex }}/${{ matrix.shardTotal }}
run: cd src/frontend/ && yarn e2e:test
- name: Save logs
if: always()
run: docker compose logs > src/frontend/apps/e2e/report/logs.txt
- name: Save Dimail logs
if: always()
run: docker compose logs dimail > src/frontend/apps/e2e/report/dimail-logs.txt
- uses: actions/upload-artifact@v4
- uses: actions/upload-artifact@v3
if: always()
with:
name: playwright-report-${{ matrix.shardIndex }}
name: playwright-report
path: src/frontend/apps/e2e/report/
retention-days: 7
tests-e2e-feedback:
runs-on: ubuntu-latest
needs: [test-e2e]
if: always()
steps:
- name: All tests ok
if: ${{ !(contains(needs.*.result, 'failure')) }}
run: exit 0
- name: Some tests failed
if: ${{ contains(needs.*.result, 'failure') }}
run: exit 1
build-mails:
runs-on: ubuntu-latest
defaults:
@@ -241,7 +162,7 @@ jobs:
working-directory: src/mail
steps:
- name: Checkout repository
uses: actions/checkout@v4
uses: actions/checkout@v2
- name: Install Node.js
uses: actions/setup-node@v4
with:
@@ -252,11 +173,6 @@ jobs:
run: yarn install --frozen-lockfile
- name: Build mails
run: yarn build
- name: Persist mails' templates
uses: actions/upload-artifact@v4
with:
name: mails-templates
path: src/backend/core/templates/mail
lint-back:
runs-on: ubuntu-latest
@@ -265,9 +181,9 @@ jobs:
working-directory: src/backend
steps:
- name: Checkout repository
uses: actions/checkout@v4
uses: actions/checkout@v2
- name: Install Python
uses: actions/setup-python@v5
uses: actions/setup-python@v3
with:
python-version: '3.10'
- name: Install development dependencies
@@ -281,7 +197,6 @@ jobs:
test-back:
runs-on: ubuntu-latest
needs: build-mails
defaults:
run:
working-directory: src/backend
@@ -311,18 +226,13 @@ jobs:
steps:
- name: Checkout repository
uses: actions/checkout@v4
uses: actions/checkout@v2
- name: Create writable /data
run: |
sudo mkdir -p /data/media && \
sudo mkdir -p /data/static
- name: Download mails' templates
uses: actions/download-artifact@v4
with:
name: mails-templates
path: src/backend/core/templates/mail
- name: Install Python
uses: actions/setup-python@v5
uses: actions/setup-python@v3
with:
python-version: '3.10'
- name: Install development dependencies
@@ -335,3 +245,67 @@ jobs:
run: python manage.py compilemessages
- name: Run tests
run: ~/.local/bin/pytest -n 2
i18n-crowdin:
runs-on: ubuntu-latest
steps:
- name: Checkout repository
uses: actions/checkout@v2
- name: Install gettext (required to make messages)
run: |
sudo apt-get update
sudo apt-get install -y gettext
- name: Install Python
uses: actions/setup-python@v3
with:
python-version: '3.10'
- name: Install development dependencies
working-directory: src/backend
run: pip install --user .[dev]
- name: Generate the translation base file
run: ~/.local/bin/django-admin makemessages --keep-pot --all
- name: Load sops secrets
uses: rouja/actions-sops@main
with:
secret-file: .github/workflows/secrets.enc.env
age-key: ${{ secrets.SOPS_PRIVATE }}
- name: Setup Node.js
uses: actions/setup-node@v4
with:
node-version: '18.x'
cache: 'yarn'
cache-dependency-path: src/frontend/yarn.lock
- name: Install dependencies
run: cd src/frontend/ && yarn install --frozen-lockfile
- name: Download sources from Crowdin to stay synchronized
run: |
docker run \
--rm \
-e CROWDIN_API_TOKEN=$CROWDIN_API_TOKEN \
-e CROWDIN_PROJECT_ID=$CROWDIN_PROJECT_ID \
-e CROWDIN_BASE_PATH=$CROWDIN_BASE_PATH \
-v "${{ github.workspace }}:/app" \
crowdin/cli:3.16.0 \
crowdin download sources -c /app/crowdin/config.yml
- name: Extract the frontend translation
run: make frontend-i18n-extract
- name: Upload files to Crowdin
run: |
docker run \
--rm \
-e CROWDIN_API_TOKEN=$CROWDIN_API_TOKEN \
-e CROWDIN_PROJECT_ID=$CROWDIN_PROJECT_ID \
-e CROWDIN_BASE_PATH=$CROWDIN_BASE_PATH \
-v "${{ github.workspace }}:/app" \
crowdin/cli:3.16.0 \
crowdin upload sources -c /app/crowdin/config.yml
-35
View File
@@ -1,35 +0,0 @@
name: Release Chart
run-name: Release Chart
on:
push:
branches:
- 'main'
paths:
- ./src/helm/desk/**
jobs:
release:
permissions:
contents: write
runs-on: ubuntu-latest
steps:
- name: Checkout
uses: actions/checkout@v4
with:
fetch-depth: 0
- name: Cleanup
run: rm -rf ./src/helm/extra
- name: Install Helm
uses: azure/setup-helm@v4
env:
GITHUB_TOKEN: "${{ secrets.GITHUB_TOKEN }}"
- name: Publish Helm charts
uses: numerique-gouv/helm-gh-pages@add-overwrite-option
with:
charts_dir: ./src/helm
linting: on
token: ${{ secrets.GITHUB_TOKEN }}
+20
View File
@@ -0,0 +1,20 @@
SOPS_PRIVATE=ENC[AES256_GCM,data:uMsnbpi1FFxO430iptp2axlH/souCPCJ/afCqh/kIDWs8xWHu0xJ2o6PlNOgb4l1gHF8sy4eSHFOW5HPKVbZ9gafRIL0JYHJr3A=,iv:IgaHDad7IuW2wFoxGALLDCs+UiSd3rEYwGNSX38wUfI=,tag:zs4wn4hDm5fghfI/7iH6CQ==,type:str]
CROWDIN_API_TOKEN=ENC[AES256_GCM,data:tZRGFs792GjKYGit+oNubCwPbjWarhlgcyQ7oStO8K3nao0lEzLrm3+ARaOEXkzEkKSal2N1c+vlYIjRyzV1X7WZ7nZQWBiihEJgAfF3NGM=,iv:hohak/1niu5kV/W4XA0KEE3UB0BuNCDxbRxb0O+Aocs=,tag:Ssi0HyEhCKb9+WFyp3/yBQ==,type:str]
CROWDIN_BASE_PATH=ENC[AES256_GCM,data:m44KrW5xJDE=,iv:bSyKrKQ0Z1yzrNaejAUyD+n1Z9z+ci92GoyS5KiiFKI=,tag:dDtpbn+6LkPwsg+qgMNWvA==,type:str]
CROWDIN_PROJECT_ID=ENC[AES256_GCM,data:Gonh6ps7,iv:yUqPty+R060m6CYrDf7na2f6h0aRpIhRCXZoJW4ThJg=,tag:ewupoenajsc9o0Aj0r/niw==,type:str]
DOCKER_HUB_PASSWORD=ENC[AES256_GCM,data:CBjUvS/UQcqdEv8=,iv:zJKGyyRnq+zxhKkYS5h3zw/J1320zrAqpiObcwiHWsA=,tag:OoQNe+4K63KXTWNsvjRZNQ==,type:str]
DOCKER_HUB_USER=ENC[AES256_GCM,data:ivs7oeBBdA==,iv:cKAQJ071XhcRuFyeHhZkscvvz7sHrKIPJg5pt76hCXg=,tag:2VRmp1ULWvVH8NHcBifBGA==,type:str]
sops_age__list_0__map_enc=-----BEGIN AGE ENCRYPTED FILE-----\nYWdlLWVuY3J5cHRpb24ub3JnL3YxCi0+IFgyNTUxOSBIbkIyRi9jTGEzL2RZcGFS\nVkhuN2xMN1FBcm04eGJQUnpvQVRpdmFmaFF3CjdoNVhSM0lrb0lTSXAyUytSKzhG\nNERYaC80OThmUUxCSmt2U2Q0Mno3b0kKLS0tIFRjQzYyOEcwTnRScGhzaGxwaGFL\namxEUXNaeVhWUjYwbzR2Z1Uzb3JsTlEK6tyhbRmcUP4Aql49DPkrYb5tbwvK2EdA\ntvyPQo4pPit+pzgqsVgW+O8Wo4/rYLlITfuVRrOfHEaH3wmf6hziSw==\n-----END AGE ENCRYPTED FILE-----\n
sops_age__list_0__map_recipient=age15fyxdwmg5mvldtqqus87xspuws2u0cpvwheehrtvkexj4tnsqqysw6re2x
sops_age__list_1__map_enc=-----BEGIN AGE ENCRYPTED FILE-----\nYWdlLWVuY3J5cHRpb24ub3JnL3YxCi0+IFgyNTUxOSBIL2NwVjVtVmd1ZE15WmFC\nVGl5cnU1OHVUcXAxWmZCQXBJbXBlcXFUN0RzClpOVXBTek9wbVB6M2s5TWlFbUo3\nelZrT3dsK3p6cEJGTGJoSTRvaEh4NW8KLS0tIGlsL29oSEthU28xV0RERFIxTnZR\nSSs3YjdUT0NyTEtlbjlmZXVOaTd3SzgK5jFJGREfJ/HVytWsCKWFsqM5JoaFSnhv\n538KzzldzcbtWfnY+bQ6A2EBjETwOzCTuQB8axAMj0URXPI+qelKIQ==\n-----END AGE ENCRYPTED FILE-----\n
sops_age__list_1__map_recipient=age16hnlml8yv4ynwy0seer57g8qww075crd0g7nsundz3pj4wk7m3vqftszg7
sops_age__list_2__map_enc=-----BEGIN AGE ENCRYPTED FILE-----\nYWdlLWVuY3J5cHRpb24ub3JnL3YxCi0+IFgyNTUxOSBzOWtwSytMNGRmSHNQN0Zz\nV1RTWktXT1c5VjYvNTg2V0x1U3JSTFA1dmtvCnc4RHRodjVxSzcrZXJycCtJMGNy\nZ3ZwbTFDTUc4TTMxbm8rNXQ1ME90Q1UKLS0tIERPUmg1dSt5OWQwaXV0Y2FiYWdt\nVzlhcFNvUS9Nd3A5ZTYwT2lLUU5WVU0KGHSm1BQ3voKs98WiXNLO6hlqoiQmi1/F\n2RCBkE4/gOVyAmJAjOFizaF0Dhd7Ba4KS5l5QylFHloL8XtyixEhog==\n-----END AGE ENCRYPTED FILE-----\n
sops_age__list_2__map_recipient=age1g5keveae6zhn059e7cxkjqdz4v3u47ypejv9ujld65nwh6d5pd9qfm0ecv
sops_age__list_3__map_enc=-----BEGIN AGE ENCRYPTED FILE-----\nYWdlLWVuY3J5cHRpb24ub3JnL3YxCi0+IFgyNTUxOSBmYlB6cEZKK3NLbm5Ob0hi\nS3M0OFoyTHhVYVA1WmcvSHc2MUErQ2JyaUh3Cnp1NlYvU1BPUnVON29UZWUvWmEz\nOW85N013VUJ3Q1ZZUmMvTWdYclBZTUkKLS0tIG1KODdkazhTUHRPMXZXQnJFNXJY\nOFlSNjZCbkpxcm9tN1dFTkZ4K1lETFUKhiPwKEG71OlTcK/Ul1GKGayLy025vAmo\nNgQUhbqXf7KmqDAmrQNzXTsLsOpRi33l66jFSGkTsFtiXNlmjFljKg==\n-----END AGE ENCRYPTED FILE-----\n
sops_age__list_3__map_recipient=age12g6f5fse25tgrwweleh4jls3qs52hey2edh759smulwmk5lnzadslu2cp3
sops_age__list_4__map_enc=-----BEGIN AGE ENCRYPTED FILE-----\nYWdlLWVuY3J5cHRpb24ub3JnL3YxCi0+IFgyNTUxOSBJUzI2eDB0MU41YWoxcjAv\nOU8yMlowb3IybCtzcFBBTEVlUzdBczZkQlFNCk5tbnhLR2kweGNFZmx1OFgwaWZU\nTzRKZTBZamJjRFNYOUJlYmxTUVg4S2MKLS0tIDNGdlVVc0hLK1BPTUJVc0tPb1lK\ndVVWVi9GRkxwYXR1cXMrdnlsejJBeGsKKmVWvIMrBYH+UrDMkZPxN8KWnCgA6WK2\nbexMYr2AIF2QMbhck7XW2NuvAwvwjbJMfcd+cp9boe+EcC4YjdJZlw==\n-----END AGE ENCRYPTED FILE-----\n
sops_age__list_4__map_recipient=age1tl80n23wq6zxegupwn70ew0yp225ua5v4dk800x7g2w6pvlxz46qk592pa
sops_lastmodified=2024-02-05T17:45:08Z
sops_mac=ENC[AES256_GCM,data:WMsYYvGId0UlRhiw5C4TTpP/7oIIgWteGRtX7Nlo3qQwBc1LSUyN+7kblkXWD+nNG51B0xfl9E5QdXO3Ao2WskQZZEJqqkmt6wSw/ScSPcH56FoIqBhoGsPKyJOZkK0zojnCJQLniTDlISG0UiiJ4KHh8FJDWYbzAy2zFEm3Et8=,iv:zB2hyxnKiA3flsMQ7XFOT/fR9hamd6YWo3BNwce4oWM=,tag:uZ2Ia8sR5R8fmURtJ+PojA==,type:str]
sops_unencrypted_suffix=_unencrypted
sops_version=3.8.1
+6 -2
View File
@@ -42,6 +42,7 @@ env.bak/
venv.bak/
env.d/development/*
!env.d/development/*.dist
env.d/terraform
# npm
node_modules
@@ -58,9 +59,13 @@ src/frontend/tsclient
# Logs
*.log
# Terraform
.terraform
*.tfstate
*.tfstate.backup
# Test & lint
.coverage
coverage.json
.pylint.d
.pytest_cache
db.sqlite3
@@ -74,4 +79,3 @@ db.sqlite3
.vscode/
*.iml
.devcontainer/
.tool-versions
-3
View File
@@ -1,3 +0,0 @@
[submodule "secrets"]
path = secrets
url = ../secrets
+12 -10
View File
@@ -1,11 +1,13 @@
creation_rules:
- path_regex: ./*
key_groups:
- age:
- age15fyxdwmg5mvldtqqus87xspuws2u0cpvwheehrtvkexj4tnsqqysw6re2x # jacques
- age16hnlml8yv4ynwy0seer57g8qww075crd0g7nsundz3pj4wk7m3vqftszg7 # github-repo
- age1plkp8td6zzfcavjusmsfrlk54t9vn8jjxm8zaz7cmnr7kzl2nfnsd54hwg # Anthony Le-Courric
- age12g6f5fse25tgrwweleh4jls3qs52hey2edh759smulwmk5lnzadslu2cp3 # Antoine Lebaud
- age1hnhuzj96ktkhpyygvmz0x9h8mfvssz7ss6emmukags644mdhf4msajk93r # Samuel Paccoud
- age1tl80n23wq6zxegupwn70ew0yp225ua5v4dk800x7g2w6pvlxz46qk592pa # Marie Pupo Jeammet
- age1rjchule5sncn8r8gfph07muee6vzx4wqfrtldt5jjzke4vlfxy2qqplfvc # Quentin Bey
# Here we have
# - Jacques key-id: age15fyxdwmg5mvldtqqus87xspuws2u0cpvwheehrtvkexj4tnsqqysw6re2x
# - github-repo key-id: age16hnlml8yv4ynwy0seer57g8qww075crd0g7nsundz3pj4wk7m3vqftszg7
# - Anthony Le-Courric key-id: age1g5keveae6zhn059e7cxkjqdz4v3u47ypejv9ujld65nwh6d5pd9qfm0ecv
# - Antoine Lebaud key-id: age12g6f5fse25tgrwweleh4jls3qs52hey2edh759smulwmk5lnzadslu2cp3
# - Marie Pupo Jeammet key-id: age1tl80n23wq6zxegupwn70ew0yp225ua5v4dk800x7g2w6pvlxz46qk592pa
- age:
age15fyxdwmg5mvldtqqus87xspuws2u0cpvwheehrtvkexj4tnsqqysw6re2x,
age16hnlml8yv4ynwy0seer57g8qww075crd0g7nsundz3pj4wk7m3vqftszg7,
age1g5keveae6zhn059e7cxkjqdz4v3u47ypejv9ujld65nwh6d5pd9qfm0ecv,
age12g6f5fse25tgrwweleh4jls3qs52hey2edh759smulwmk5lnzadslu2cp3,
age1tl80n23wq6zxegupwn70ew0yp225ua5v4dk800x7g2w6pvlxz46qk592pa
-290
View File
@@ -7,293 +7,3 @@ and this project adheres to
[Semantic Versioning](https://semver.org/spec/v2.0.0.html).
## [Unreleased]
### Added
- ✨(api) add count mailboxes to MailDomain serializer
- ✨(dimail) manage 'action required' status for MailDomain
- ✨(domains) add action required status on MailDomain
- ✨(dimail) send pending mailboxes upon domain activation
### Fixed
- 🚑️(teams) do not display add button when disallowed #676
- 🚑️(plugins) fix name from SIRET specific case #674
- 🐛(api) restrict mailbox sync to enabled domains
## [1.10.1] - 2025-01-27
### Added
- ✨(dimail) management command to fetch domain status
### Changed
- ✨(scripts) adapts release script after moving the deployment part
### Fixed
- 🐛(dimail) fix imported mailboxes should be enabled instead of pending #659
- ⚡️(api) add missing cache for stats endpoint
## [1.10.0] - 2025-01-21
### Added
- ✨(api) create stats endpoint
- ✨(teams) add Team dependencies #560
- ✨(organization) add admin action for plugin #640
- ✨(anct) fetch and display organization names of communes #583
- ✨(frontend) display email if no username #562
- 🧑‍💻(oidc) add ability to pull registration ID (e.g. SIRET) from OIDC #577
### Fixed
- 🐛(frontend) improve e2e tests avoiding race condition from mocks #641
- 🐛(backend) fix flaky test with search contact #605
- 🐛(backend) fix flaky test with team access #646
- 🧑‍💻(dimail) remove 'NoneType: None' log in debug mode
- 🐛(frontend) fix flaky e2e test #636
- 🐛(frontend) fix disable mailbox button display #631
- 🐛(backend) fix dimail call despite mailbox creation failure on our side
- 🧑‍💻(user) fix the User.language infinite migration #611
## [1.9.1] - 2024-12-18
## [1.9.0] - 2024-12-17
### Fixed
- 🐛(backend) fix manage roles on domain admin view
### Added
- ✨(backend) add admin action to check domain health
- ✨(dimail) check domain health
- ✨(frontend) disable mailbox and allow to create pending mailbox
- ✨(organizations) add siret to name conversion #584
- 💄(frontend) redirect home according to abilities #588
- ✨(maildomain_access) add API endpoint to search users #508
## [1.8.0] - 2024-12-12
### Added
- ✨(contacts) add "abilities" to API endpoint data #585
- ✨(contacts) allow filter list API with email
- ✨(contacts) return profile contact from same organization
- ✨(dimail) automate allows requests to dimail
- ✨(contacts) add notes & force full_name #565
### Changed
- ♻️(contacts) move user profile to contact #572
- ♻️(contacts) split api test module in actions #573
### Fixed
- 🐛(backend) fix manage roles on domain admin view
- 🐛(mailbox) fix activate mailbox feature
- 🔧(helm) fix the configuration environment #579
## [1.7.1] - 2024-11-28
## [1.7.0] - 2024-11-28
### Added
- ✨(mailbox) allow to activate mailbox
- ✨(mailbox) allow to disable mailbox
- ✨(backend) add ServiceProvider #522
- 💄(admin) allow header color customization #552
- ✨(organization) add API endpoints #551
### Fixed
- 🐛(admin) add organization on user #555
## [1.6.1] - 2024-11-22
### Fixed
- 🩹(mailbox) fix status of current mailboxes
- 🚑️(backend) fix claim contains non-user field #548
## [1.6.0] - 2024-11-20
### Removed
- 🔥(teams) remove search by trigram for team access and contact
### Added
- ✨(domains) allow creation of "pending" mailboxes
- ✨(teams) allow team management for team admins/owners #509
### Fixed
- 🔊(backend) update logger config to info #542
## [1.5.0] - 2024-11-14
### Removed
- ⬆️(dependencies) remove unneeded dependencies
- 🔥(teams) remove pagination of teams listing
- 🔥(teams) remove search users by trigram
- 🗃️(teams) remove `slug` field
### Added
- ✨(dimail) send domain creation requests to dimail #454
- ✨(dimail) synchronize mailboxes from dimail to our db #453
- ✨(ci) add security scan #429
- ✨(teams) register contacts on admin views
### Fixed
- 🐛(mail) fix display button on outlook
- 💚(ci) improve E2E tests #492
- 🔧(sentry) restore default integrations
- 🔇(backend) remove Sentry duplicated warning/errors
- 👷(ci) add sharding e2e tests #467
- 🐛(dimail) fix unexpected status_code for proper debug #454
## [1.4.1] - 2024-10-23
### Fixed
- 🚑️(frontend) fix MailDomainsLayout
## [1.4.0] - 2024-10-23
### Added
- ✨(frontend) add tabs inside #466
### Fixed
- ✏️(mail) fix typo into mailbox creation email
- 🐛(sentry) fix duplicated sentry errors #479
- 🐛(script) improve and fix release script
## [1.3.1] - 2024-10-18
## [1.3.0] - 2024-10-18
### Added
- ✨(api) add RELEASE version on config endpoint #459
- ✨(backend) manage roles on domain admin view
- ✨(frontend) show version number in footer #369
- 👔(backend) add Organization model
### Changed
- 🛂(backend) match email if no existing user matches the sub
### Fixed
- 💄(mail) improve mailbox creation email #462
- 🐛(frontend) fix update accesses form #448
- 🛂(backend) do not duplicate user when disabled
## [1.2.1] - 2024-10-03
### Fixed
- 🔧(mail) use new scaleway email gateway #435
## [1.2.0] - 2024-09-30
### Added
- ✨(ci) add helmfile linter and fix argocd sync #424
- ✨(domains) add endpoint to list and retrieve domain accesses #404
- 🍱(dev) embark dimail-api as container #366
- ✨(dimail) allow la regie to request a token for another user #416
- ✨(frontend) show username on AccountDropDown #412
- 🥅(frontend) improve add & update group forms error handling #387
- ✨(frontend) allow group members filtering #363
- ✨(mailbox) send new mailbox confirmation email #397
- ✨(domains) domain accesses update API #423
- ✨(backend) domain accesses create API #428
- 🥅(frontend) catch new errors on mailbox creation #392
- ✨(api) domain accesses delete API #433
- ✨(frontend) add mail domain access management #413
### Fixed
- ♿️(frontend) fix left nav panel #396
- 🔧(backend) fix configuration to avoid different ssl warning #432
### Changed
- ♻️(serializers) move business logic to serializers #414
## [1.1.0] - 2024-09-10
### Added
- 📈(monitoring) configure sentry monitoring #378
- 🥅(frontend) improve api error handling #355
### Changed
- 🗃️(models) move dimail 'secret' to settings #372
### Fixed
- 🐛(dimail) improve handling of dimail errors on failed mailbox creation #377
- 💬(frontend) fix group member removal text #382
- 💬(frontend) fix add mail domain text #382
- 🐛(frontend) fix keyboard navigation #379
- 🐛(frontend) fix add mail domain form submission #355
## [1.0.2] - 2024-08-30
### Added
- 🔧Runtime config for the frontend (#345)
- 🔧(helm) configure resource server in staging
### Fixed
- 👽️(mailboxes) fix mailbox creation after dimail api improvement (#360)
## [1.0.1] - 2024-08-19
### Fixed
- ✨(frontend) user can add mail domains
## [1.0.0] - 2024-08-09
### Added
- ✨(domains) create and manage domains on admin + API
- ✨(domains) mailbox creation + link to email provisioning API
[unreleased]: https://github.com/numerique-gouv/people/compare/v1.10.1...main
[1.10.1]: https://github.com/numerique-gouv/people/releases/v1.10.1
[1.10.0]: https://github.com/numerique-gouv/people/releases/v1.10.0
[1.9.1]: https://github.com/numerique-gouv/people/releases/v1.9.1
[1.9.0]: https://github.com/numerique-gouv/people/releases/v1.9.0
[1.8.0]: https://github.com/numerique-gouv/people/releases/v1.8.0
[1.7.1]: https://github.com/numerique-gouv/people/releases/v1.7.1
[1.7.0]: https://github.com/numerique-gouv/people/releases/v1.7.0
[1.6.1]: https://github.com/numerique-gouv/people/releases/v1.6.1
[1.6.0]: https://github.com/numerique-gouv/people/releases/v1.6.0
[1.5.0]: https://github.com/numerique-gouv/people/releases/v1.5.0
[1.4.1]: https://github.com/numerique-gouv/people/releases/v1.4.1
[1.4.0]: https://github.com/numerique-gouv/people/releases/v1.4.0
[1.3.1]: https://github.com/numerique-gouv/people/releases/v1.3.1
[1.3.0]: https://github.com/numerique-gouv/people/releases/v1.3.0
[1.2.1]: https://github.com/numerique-gouv/people/releases/v1.2.1
[1.2.0]: https://github.com/numerique-gouv/people/releases/v1.2.0
[1.1.0]: https://github.com/numerique-gouv/people/releases/v1.1.0
[1.0.2]: https://github.com/numerique-gouv/people/releases/v1.0.2
[1.0.1]: https://github.com/numerique-gouv/people/releases/v1.0.1
[1.0.0]: https://github.com/numerique-gouv/people/releases/v1.0.0
+31 -72
View File
@@ -1,66 +1,18 @@
# Django People
# ---- base image to inherit from ----
FROM python:3.12.6-alpine3.20 AS base
FROM python:3.10-slim-bullseye as base
# Upgrade pip to its latest release to speed up dependencies installation
RUN python -m pip install --upgrade pip setuptools
RUN python -m pip install --upgrade pip
# Upgrade system packages to install security updates
RUN apk update && \
apk upgrade
### ---- Front-end dependencies image ----
FROM node:20 AS frontend-deps
WORKDIR /deps
COPY ./src/frontend/package.json ./package.json
COPY ./src/frontend/yarn.lock ./yarn.lock
COPY ./src/frontend/apps/desk/package.json ./apps/desk/package.json
COPY ./src/frontend/packages/i18n/package.json ./packages/i18n/package.json
COPY ./src/frontend/packages/eslint-config-people/package.json ./packages/eslint-config-people/package.json
RUN yarn --frozen-lockfile
### ---- Front-end builder dev image ----
FROM node:20 AS frontend-builder-dev
WORKDIR /builder
COPY --from=frontend-deps /deps/node_modules ./node_modules
COPY ./src/frontend .
WORKDIR ./apps/desk
### ---- Front-end builder image ----
FROM frontend-builder-dev AS frontend-builder
RUN yarn build
# ---- Front-end image ----
FROM nginxinc/nginx-unprivileged:1.26-alpine AS frontend-production
# Un-privileged user running the application
ARG DOCKER_USER
USER ${DOCKER_USER}
COPY --from=frontend-builder \
/builder/apps/desk/out \
/usr/share/nginx/html
COPY ./src/frontend/apps/desk/conf/default.conf /etc/nginx/conf.d
# Copy entrypoint
COPY ./docker/files/usr/local/bin/entrypoint /usr/local/bin/entrypoint
ENTRYPOINT [ "/usr/local/bin/entrypoint" ]
CMD ["nginx", "-g", "daemon off;"]
RUN apt-get update && \
apt-get -y upgrade && \
rm -rf /var/lib/apt/lists/*
# ---- Back-end builder image ----
FROM base AS back-builder
FROM base as back-builder
WORKDIR /builder
@@ -72,7 +24,7 @@ RUN mkdir /install && \
# ---- mails ----
FROM node:20 AS mail-builder
FROM node:18 as mail-builder
COPY ./src/mail /mail/app
@@ -83,13 +35,15 @@ RUN yarn install --frozen-lockfile && \
# ---- static link collector ----
FROM base AS link-collector
FROM base as link-collector
ARG PEOPLE_STATIC_ROOT=/data/static
# Install libpangocairo & rdfind
RUN apk add \
pango \
rdfind
RUN apt-get update && \
apt-get install -y \
libpangocairo-1.0-0 \
rdfind && \
rm -rf /var/lib/apt/lists/*
# Copy installed python dependencies
COPY --from=back-builder /install /usr/local
@@ -108,18 +62,21 @@ RUN DJANGO_CONFIGURATION=Build DJANGO_JWT_PRIVATE_SIGNING_KEY=Dummy \
RUN rdfind -makesymlinks true -followsymlinks true -makeresultsfile false ${PEOPLE_STATIC_ROOT}
# ---- Core application image ----
FROM base AS core
FROM base as core
ENV PYTHONUNBUFFERED=1
# Install required system libs
RUN apk add \
gettext \
cairo \
libffi-dev \
gdk-pixbuf \
pango \
shared-mime-info
RUN apt-get update && \
apt-get install -y \
gettext \
libcairo2 \
libffi-dev \
libgdk-pixbuf2.0-0 \
libpango-1.0-0 \
libpangocairo-1.0-0 \
shared-mime-info && \
rm -rf /var/lib/apt/lists/*
# Copy entrypoint
COPY ./docker/files/usr/local/bin/entrypoint /usr/local/bin/entrypoint
@@ -143,13 +100,15 @@ WORKDIR /app
ENTRYPOINT [ "/usr/local/bin/entrypoint" ]
# ---- Development image ----
FROM core AS backend-development
FROM core as development
# Switch back to the root user to install development dependencies
USER root:root
# Install psql
RUN apk add postgresql-client
RUN apt-get update && \
apt-get install -y postgresql-client && \
rm -rf /var/lib/apt/lists/*
# Uninstall people and re-install it in editable mode along with development
# dependencies
@@ -166,10 +125,10 @@ ENV DB_HOST=postgresql \
DB_PORT=5432
# Run django development server
CMD ["python", "manage.py", "runserver", "0.0.0.0:8000"]
CMD python manage.py runserver 0.0.0.0:8000
# ---- Production image ----
FROM core AS backend-production
FROM core as production
ARG PEOPLE_STATIC_ROOT=/data/static
@@ -188,4 +147,4 @@ COPY --from=link-collector ${PEOPLE_STATIC_ROOT} ${PEOPLE_STATIC_ROOT}
COPY --from=mail-builder /mail/backend/core/templates/mail /app/core/templates/mail
# The default command runs gunicorn WSGI server in people's main module
CMD ["gunicorn", "-c", "/usr/local/etc/gunicorn/people.py", "people.wsgi:application"]
CMD gunicorn -c /usr/local/etc/gunicorn/people.py people.wsgi:application
+4 -89
View File
@@ -88,7 +88,6 @@ bootstrap: \
back-i18n-compile \
mails-install \
mails-build \
dimail-setup-db \
install-front-desk
.PHONY: bootstrap
@@ -110,7 +109,6 @@ run: ## start the wsgi (production) and development server
@$(COMPOSE) up --force-recreate -d app-dev
@$(COMPOSE) up --force-recreate -d celery-dev
@$(COMPOSE) up --force-recreate -d keycloak
@$(COMPOSE) up -d dimail
@echo "Wait for postgresql to be up..."
@$(WAIT_KC_DB)
@$(WAIT_DB)
@@ -131,7 +129,6 @@ demo: ## flush db then create a demo for load testing purpose
@$(MANAGE) create_demo
.PHONY: demo
# Nota bene: Black should come after isort just in case they don't agree...
lint: ## lint back-end python sources
lint: \
@@ -169,35 +166,24 @@ test-back-parallel: ## run all back-end tests in parallel
bin/pytest -n auto $${args:-${1}}
.PHONY: test-back-parallel
test-coverage: ## compute, display and save test coverage
bin/pytest --cov=. --cov-report json .
.PHONY: test-coverage
makemigrations: ## run django makemigrations for the people project.
@echo "$(BOLD)Running makemigrations$(RESET)"
@$(COMPOSE) up -d postgresql
@$(WAIT_DB)
@$(MANAGE) makemigrations $(ARGS)
@$(MANAGE) makemigrations
.PHONY: makemigrations
migrate: ## run django migrations for the people project.
@echo "$(BOLD)Running migrations$(RESET)"
@$(COMPOSE) up -d postgresql
@$(WAIT_DB)
@$(MANAGE) migrate $(ARGS)
@$(MANAGE) migrate
.PHONY: migrate
showmigrations: ## run django showmigrations for the people project.
@echo "$(BOLD)Running showmigrations$(RESET)"
@$(COMPOSE) up -d postgresql
@$(WAIT_DB)
@$(MANAGE) showmigrations $(ARGS)
.PHONY: showmigrations
superuser: ## Create an admin superuser with password "admin"
@echo "$(BOLD)Creating a Django superuser$(RESET)"
$(MANAGE) createsuperuser --username admin --password admin
@$(MANAGE) createsuperuser --email admin@example.com --password admin
.PHONY: superuser
back-i18n-compile: ## compile the gettext files
@@ -218,10 +204,9 @@ dbshell: ## connect to database shell
docker compose exec app-dev python manage.py dbshell
.PHONY: dbshell
resetdb: FLUSH_ARGS ?=
resetdb: ## flush database and create a superuser "admin"
@echo "$(BOLD)Flush database$(RESET)"
@$(MANAGE) flush $(FLUSH_ARGS)
@$(MANAGE) flush
@${MAKE} superuser
.PHONY: resetdb
@@ -275,17 +260,6 @@ i18n-generate-and-upload: \
crowdin-upload
.PHONY: i18n-generate-and-upload
# -- INTEROPERABILTY
# -- Dimail configuration
recreate-dimail-container:
@$(COMPOSE) up --force-recreate -d dimail
.PHONY: recreate-dimail-container
dimail-setup-db:
@echo "$(BOLD)Populating database of local dimail API container$(RESET)"
@$(MANAGE) setup_dimail_db
.PHONY: dimail-setup-db
# -- Mail generator
@@ -348,62 +322,3 @@ frontend-i18n-generate: \
frontend-i18n-compile: ## Format the crowin json files used deploy to the apps
cd $(PATH_FRONT) && yarn i18n:deploy
.PHONY: frontend-i18n-compile
# -- K8S
start-kind: ## Create the kubernetes cluster
./bin/start-kind.sh
.PHONY: start-kind
install-external-secrets: ## install the kubernetes secrets from Vaultwarden
./bin/install-external-secrets.sh
.PHONY: build-k8s-cluster
tilt-up: ## start tilt - k8s local development
tilt up -f ./bin/Tiltfile
.PHONY: tilt-up
start-tilt-keycloak: ## start the kubernetes cluster using kind, without Pro Connect for authentication, use keycloak
DEV_ENV=dev-keycloak tilt up -f ./bin/Tiltfile
.PHONY: build-k8s-cluster
release: ## helper for release and deployment
python scripts/release.py
.PHONY: release
install-secret: ## install the kubernetes secrets from Vaultwarden
if kubectl -n desk get secrets bitwarden-cli-desk; then \
echo "Secret already present"; \
else \
echo "Please provide the following information:"; \
read -p "Enter your vaultwarden email login: " LOGIN; \
read -p "Enter your vaultwarden password: " PASSWORD; \
read -p "Enter your vaultwarden server url: " URL; \
echo "\nCreate vaultwarden secret"; \
echo "apiVersion: v1" > /tmp/secret.yaml; \
echo "kind: Secret" >> /tmp/secret.yaml; \
echo "metadata:" >> /tmp/secret.yaml; \
echo " name: bitwarden-cli-desk" >> /tmp/secret.yaml; \
echo " namespace: desk" >> /tmp/secret.yaml; \
echo "type: Opaque" >> /tmp/secret.yaml; \
echo "stringData:" >> /tmp/secret.yaml; \
echo " BW_HOST: $$URL" >> /tmp/secret.yaml; \
echo " BW_PASSWORD: $$PASSWORD" >> /tmp/secret.yaml; \
echo " BW_USERNAME: $$LOGIN" >> /tmp/secret.yaml; \
kubectl -n desk apply -f /tmp/secret.yaml;\
rm -f /tmp/secret.yaml; \
fi; \
if kubectl get ns external-secrets; then \
echo "External secret already deployed"; \
else \
helm repo add external-secrets https://charts.external-secrets.io; \
helm upgrade --install external-secrets \
external-secrets/external-secrets \
-n external-secrets \
--create-namespace \
--set installCRDs=true; \
fi
.PHONY: build-k8s-cluster
fetch-domain-status:
@$(MANAGE) fetch_domain_status
.PHONY: fetch-domain-status
+13 -38
View File
@@ -1,12 +1,12 @@
# People
People is an application to handle users and teams, and distribute permissions accross [La Suite](https://lasuite.numerique.gouv.fr/).
People is an application to handle users and teams.
It is built on top of [Django Rest
As of today, this project is **not yet ready for production**. Expect breaking changes.
People is built on top of [Django Rest
Framework](https://www.django-rest-framework.org/).
All interoperabilities will be described in `docs/interoperability`.
## Getting started
### Prerequisite
@@ -25,7 +25,7 @@ $ docker compose -v
> ⚠️ You may need to run the following commands with `sudo` but this can be
> avoided by assigning your user to the `docker` group.
### Bootstrap project
### Project bootstrap
The easiest way to start working on the project is to use GNU Make:
@@ -38,7 +38,7 @@ database migrations and compile translations. It's a good idea to use this
command each time you are pulling code from the project repository to avoid
dependency-related or migration-related issues.
Your Docker services should now be up and running! 🎉
Your Docker services should now be up and running 🎉
Note that if you need to run them afterward, you can use the eponym Make rule:
@@ -46,7 +46,13 @@ Note that if you need to run them afterward, you can use the eponym Make rule:
$ make run
```
You can check all available Make rules using:
### Adding content
You can create a basic demo site by running:
$ make demo
Finally, you can check all available Make rules using:
```bash
$ make help
@@ -63,37 +69,6 @@ You first need to create a superuser account:
$ make superuser
```
You can then login with sub `admin` and password `admin`.
### Adding demo content
You can create a basic demo site by running:
```bash
$ make demo
```
### Setting dimail database
To ease local development when working on interoperability between people and dimail, we embark dimail-api in a container running in "fake" mode.
To populate dimail local database with users/domains/permissions needed for basic development:
- log in with "people" user
- run `make dimail-setup-db`
### Run frontend
Run the front with:
```bash
$ make run-front-desk
```
Then access [http://localhost:3000](http://localhost:3000) with :
user: people
password: people
## Contributing
This project is intended to be community-driven, so please, do not hesitate to
-74
View File
@@ -1,74 +0,0 @@
load('ext://uibutton', 'cmd_button', 'bool_input', 'location')
load('ext://namespace', 'namespace_create', 'namespace_inject')
namespace_create('desk')
docker_build(
'localhost:5001/people-backend:latest',
context='..',
dockerfile='../Dockerfile',
only=['./src/backend', './src/mail', './docker'],
target = 'backend-production',
live_update=[
sync('../src/backend', '/app'),
run(
'pip install -r /app/requirements.txt',
trigger=['./api/requirements.txt']
)
]
)
docker_build(
'localhost:5001/people-frontend:latest',
context='..',
dockerfile='../Dockerfile',
build_args={'ENV': 'dev'},
only=['./src/frontend', './src/mail', './docker'],
target = 'frontend-builder-dev',
live_update=[
sync('../src/frontend', '/builder'),
]
)
# helmfile in docker mount the current working directory and the helmfile.yaml
# requires the keycloak config in another directory
k8s_yaml(local('cd .. && helmfile -n desk -e ${DEV_ENV:-dev} template --file ./src/helm/helmfile.yaml'))
migration = '''
set -eu
# get k8s pod name from tilt resource name
POD_NAME="$(tilt get kubernetesdiscovery desk-backend -ojsonpath='{.status.pods[0].name}')"
kubectl -n desk exec "$POD_NAME" -- python manage.py makemigrations
'''
cmd_button('Make migration',
argv=['sh', '-c', migration],
resource='desk-backend',
icon_name='developer_board',
text='Run makemigration',
)
pod_migrate = '''
set -eu
# get k8s pod name from tilt resource name
POD_NAME="$(tilt get kubernetesdiscovery desk-backend -ojsonpath='{.status.pods[0].name}')"
kubectl -n desk exec "$POD_NAME" -- python manage.py migrate --no-input
'''
cmd_button('Migrate db',
argv=['sh', '-c', pod_migrate],
resource='desk-backend',
icon_name='developer_board',
text='Run database migration',
)
# Command to created domain/users/access from people to dimail
populate_dimail_from_people = '''
set -eu
# get k8s pod name from tilt resource name
POD_NAME="$(tilt get kubernetesdiscovery desk-backend -ojsonpath='{.status.pods[0].name}')"
kubectl -n desk exec "$POD_NAME" -- python manage.py setup_dimail_db --populate-from-people
'''
cmd_button('Populate dimail from people',
argv=['sh', '-c', populate_dimail_from_people],
resource='desk-backend',
icon_name='developer_board',
text='Populate dimail from people',
)
+64
View File
@@ -5,6 +5,7 @@ set -eo pipefail
REPO_DIR="$(cd "$( dirname "${BASH_SOURCE[0]}" )/.." && pwd)"
UNSET_USER=0
TERRAFORM_DIRECTORY="./env.d/terraform"
COMPOSE_FILE="${REPO_DIR}/docker-compose.yml"
COMPOSE_PROJECT="people"
@@ -91,3 +92,66 @@ function _dc_exec() {
function _django_manage() {
_dc_run "app-dev" python manage.py "$@"
}
# _set_openstack_project: select an OpenStack project from the openrc files defined in the
# terraform directory.
#
# usage: _set_openstack_project
#
# If necessary the script will prompt the user to choose a project from those available
function _set_openstack_project() {
declare prompt
declare -a projects
declare -i default=1
declare -i choice=0
declare -i n_projects
# List projects by looking in the "./env.d/terraform" directory
# and store them in an array
read -r -a projects <<< "$(
find "${TERRAFORM_DIRECTORY}" -maxdepth 1 -mindepth 1 -type d |
sed 's|'"${TERRAFORM_DIRECTORY}\/"'||' |
xargs
)"
nb_projects=${#projects[@]}
if [[ ${nb_projects} -le 0 ]]; then
echo "There are no OpenStack projects defined..." >&2
echo "To add one, create a subdirectory in \"${TERRAFORM_DIRECTORY}\" with the name" \
"of your project and copy your \"openrc.sh\" file into it." >&2
exit 10
fi
if [[ ${nb_projects} -gt 1 ]]; then
prompt="Select an OpenStack project to target:\\n"
for (( i=0; i<nb_projects; i++ )); do
prompt+="[$((i+1))] ${projects[$i]}"
if [[ $((i+1)) -eq ${default} ]]; then
prompt+=" (default)"
fi
prompt+="\\n"
done
prompt+="If your OpenStack project is not listed, add it to the \"env.d/terraform\" directory.\\n"
prompt+="Your choice: "
read -r -p "$(echo -e "${prompt}")" choice
if [[ ${choice} -gt nb_projects ]]; then
(>&2 echo "Invalid choice ${choice} (should be <= ${nb_projects})")
exit 11
fi
if [[ ${choice} -le 0 ]]; then
choice=${default}
fi
fi
project=${projects[$((choice-1))]}
# Check that the openrc.sh file exists for this project
if [ ! -f "${TERRAFORM_DIRECTORY}/${project}/openrc.sh" ]; then
(>&2 echo "Missing \"openrc.sh\" file in \"${TERRAFORM_DIRECTORY}/${project}\". Check documentation.")
exit 12
fi
echo "${project}"
}
-90
View File
@@ -1,90 +0,0 @@
#!/usr/bin/env bash
set -o errexit
CURRENT_DIR=$(pwd)
NAMESPACE=${1:-desk}
SECRET_NAME=${2:-bitwarden-cli-desk}
TEMP_SECRET_FILE=$(mktemp)
cleanup() {
rm -f "${TEMP_SECRET_FILE}"
}
trap cleanup EXIT
# Check if kubectl is available
check_prerequisites() {
if ! command -v kubectl &> /dev/null; then
echo "Error: kubectl is not installed or not in PATH"
exit 1
fi
}
# Check if secret already exists
check_secret_exists() {
kubectl -n "${NAMESPACE}" get secrets "${SECRET_NAME}" &> /dev/null
}
# Collect user input securely
get_user_input() {
echo "Please provide the following information:"
read -p "Enter your Vaultwarden email login: " LOGIN
read -s -p "Enter your Vaultwarden password: " PASSWORD
echo
read -p "Enter your Vaultwarden server url: " URL
}
# Create and apply the secret
create_secret() {
cat > "${TEMP_SECRET_FILE}" << EOF
apiVersion: v1
kind: Secret
metadata:
name: ${SECRET_NAME}
namespace: ${NAMESPACE}
type: Opaque
stringData:
BW_HOST: ${URL}
BW_PASSWORD: ${PASSWORD}
BW_USERNAME: ${LOGIN}
EOF
kubectl -n "${NAMESPACE}" apply -f "${TEMP_SECRET_FILE}"
}
# Install external-secrets using Helm
install_external_secrets() {
if ! kubectl get ns external-secrets &>/dev/null; then
echo "Installing external-secrets…"
helm repo add external-secrets https://charts.external-secrets.io
helm upgrade --install external-secrets \
external-secrets/external-secrets \
-n external-secrets \
--create-namespace \
--set installCRDs=true
else
echo "External secrets already deployed"
fi
}
main() {
check_prerequisites
if check_secret_exists; then
echo "Secret '${SECRET_NAME}' already present in namespace '${NAMESPACE}'"
exit 0
fi
echo -e ${TEMP_SECRET_FILE}
get_user_input
echo -e "\nCreating Vaultwarden secret…"
create_secret
install_external_secrets
echo "Secret installation completed successfully"
}
main "$@"
-3
View File
@@ -1,3 +0,0 @@
#!/usr/bin/env bash
curl https://raw.githubusercontent.com/numerique-gouv/tools/refs/heads/main/kind/create_cluster.sh | bash -s -- desk
Executable
+25
View File
@@ -0,0 +1,25 @@
#!/usr/bin/env bash
set -eo pipefail
source "$(dirname "${BASH_SOURCE[0]}")/_config.sh"
project=$(_set_openstack_project)
echo "Using \"${project}\" project..."
source "${TERRAFORM_DIRECTORY}/${project}/openrc.sh"
# Run Terraform commands in the Hashicorp docker container via docker compose
# shellcheck disable=SC2068
DOCKER_USER="$(id -u):$(id -g)" \
PROJECT="${project}" \
docker compose run --rm \
-e OS_AUTH_URL \
-e OS_IDENTITY_API_VERSION \
-e OS_USER_DOMAIN_NAME \
-e OS_PROJECT_DOMAIN_NAME \
-e OS_TENANT_ID \
-e OS_TENANT_NAME \
-e OS_USERNAME \
-e OS_PASSWORD \
-e OS_REGION_NAME \
terraform-state "$@"
Executable
+26
View File
@@ -0,0 +1,26 @@
#!/usr/bin/env bash
set -eo pipefail
source "$(dirname "${BASH_SOURCE[0]}")/_config.sh"
project=$(_set_openstack_project)
echo "Using \"${project}\" project..."
source "${TERRAFORM_DIRECTORY}/${project}/openrc.sh"
# Run Terraform commands in the Hashicorp docker container via docker compose
# shellcheck disable=SC2068
DOCKER_USER="$(id -u):$(id -g)" \
PROJECT="${project}" \
docker compose run --rm \
-e OS_AUTH_URL \
-e OS_IDENTITY_API_VERSION \
-e OS_USER_DOMAIN_NAME \
-e OS_PROJECT_DOMAIN_NAME \
-e OS_TENANT_ID \
-e OS_TENANT_NAME \
-e OS_USERNAME \
-e OS_PASSWORD \
-e OS_REGION_NAME \
-e TF_VAR_user_name \
terraform "$@"
+29 -17
View File
@@ -1,3 +1,5 @@
version: '3.8'
services:
postgresql:
image: postgres:16
@@ -17,11 +19,11 @@ services:
app-dev:
build:
context: .
target: backend-development
target: development
args:
DOCKER_USER: ${DOCKER_USER:-1000}
user: ${DOCKER_USER:-1000}
image: people:backend-development
image: people:development
environment:
- PYLINTHOME=/app/.pylint.d
- DJANGO_CONFIGURATION=Development
@@ -35,14 +37,13 @@ services:
- ./data/media:/data/media
- ./data/static:/data/static
depends_on:
- dimail
- postgresql
- mailcatcher
- redis
celery-dev:
user: ${DOCKER_USER:-1000}
image: people:backend-development
image: people:development
command: ["celery", "-A", "people.celery_app", "worker", "-l", "DEBUG"]
environment:
- DJANGO_CONFIGURATION=Development
@@ -59,11 +60,11 @@ services:
app:
build:
context: .
target: backend-production
target: production
args:
DOCKER_USER: ${DOCKER_USER:-1000}
user: ${DOCKER_USER:-1000}
image: people:backend-production
image: people:production
environment:
- DJANGO_CONFIGURATION=Demo
env_file:
@@ -77,7 +78,7 @@ services:
celery:
user: ${DOCKER_USER:-1000}
image: people:backend-production
image: people:production
command: ["celery", "-A", "people.celery_app", "worker", "-l", "INFO"]
environment:
- DJANGO_CONFIGURATION=Demo
@@ -90,9 +91,13 @@ services:
nginx:
image: nginx:1.25
ports:
- "8082:8082"
- "8083:8083"
- "8088:8088"
volumes:
- ./docker/files/etc/nginx/conf.d:/etc/nginx/conf.d:ro
- ./src/frontend/apps/desk/out:/home/desk
- ./data/media:/data/media:ro
depends_on:
- app
- keycloak
@@ -117,8 +122,25 @@ services:
volumes:
- ".:/app"
terraform-state:
image: hashicorp/terraform:1.6
environment:
- TF_WORKSPACE=${PROJECT:-} # avoid env conflict in local state
user: ${DOCKER_USER:-1000}
working_dir: /app
volumes:
- ./src/terraform/create_state_bucket:/app
terraform:
image: hashicorp/terraform:1.6
user: ${DOCKER_USER:-1000}
working_dir: /app
volumes:
- ./src/terraform:/app
kc_postgresql:
image: postgres:14.3
platform: linux/amd64
ports:
- "5433:5432"
env_file:
@@ -151,13 +173,3 @@ services:
- "8080:8080"
depends_on:
- kc_postgresql
dimail:
entrypoint: /opt/dimail-api/start-dev.sh
image: registry.mim-libre.fr/dimail/dimail-api:v0.1.1
pull_policy: always
environment:
DIMAIL_MODE: FAKE
DIMAIL_JWT_SECRET: fake_jwt_secret
ports:
- "8001:8000"
+2 -268
View File
@@ -58,23 +58,6 @@
],
"realmRoles": ["user"]
},
{
"username": "e2e.marie",
"email": "marie.varzy@gmail.com",
"firstName": "Marie",
"lastName": "Delamairie",
"enabled": true,
"attributes": {
"siret": "21580304000017"
},
"credentials": [
{
"type": "password",
"value": "password-e2e.marie"
}
],
"realmRoles": ["user"]
},
{
"username": "user-e2e-chromium",
"email": "user@chromium.e2e",
@@ -116,216 +99,6 @@
}
],
"realmRoles": ["user"]
},
{
"username": "e2e.team-member",
"email": "e2e.team-member@example.com",
"firstName": "E2E",
"lastName": "Group Member",
"enabled": true,
"credentials": [
{
"type": "password",
"value": "password-e2e.team-member"
}
],
"realmRoles": ["user"]
},
{
"username": "e2e.team-administrator",
"email": "e2e.team-administrator@example.com",
"firstName": "E2E",
"lastName": "Group Administrator",
"enabled": true,
"credentials": [
{
"type": "password",
"value": "password-e2e.team-administrator"
}
],
"realmRoles": ["user"]
},
{
"username": "e2e.team-owner",
"email": "e2e.team-owner@example.com",
"firstName": "E2E",
"lastName": "Group Owner",
"enabled": true,
"credentials": [
{
"type": "password",
"value": "password-e2e.team-owner"
}
],
"realmRoles": ["user"]
},
{
"username": "e2e.mail-member",
"email": "e2e.mail-member@example.com",
"firstName": "E2E",
"lastName": "Mailbox Member",
"enabled": true,
"credentials": [
{
"type": "password",
"value": "password-e2e.mail-member"
}
],
"realmRoles": ["user"]
},
{
"username": "e2e.mail-administrator",
"email": "e2e.mail-administrator@example.com",
"firstName": "E2E",
"lastName": "Mailbox Administrator",
"enabled": true,
"credentials": [
{
"type": "password",
"value": "password-e2e.mail-administrator"
}
],
"realmRoles": ["user"]
},
{
"username": "e2e.mail-owner",
"email": "e2e.mail-owner@example.com",
"firstName": "E2E",
"lastName": "Mailbox Owner",
"enabled": true,
"credentials": [
{
"type": "password",
"value": "password-e2e.mail-owner"
}
],
"realmRoles": ["user"]
},
{
"username": "e2e.team-member-mail-member",
"email": "e2e.team-member-mail-member@example.com",
"firstName": "E2E",
"lastName": "Group Member Mailbox Member",
"enabled": true,
"credentials": [
{
"type": "password",
"value": "password-e2e.team-member-mail-member"
}
],
"realmRoles": ["user"]
},
{
"username": "e2e.team-member-mail-administrator",
"email": "e2e.team-member-mail-administrator@example.com",
"firstName": "E2E",
"lastName": "Group Member Mailbox Administrator",
"enabled": true,
"credentials": [
{
"type": "password",
"value": "password-e2e.team-member-mail-administrator"
}
],
"realmRoles": ["user"]
},
{
"username": "e2e.team-member-mail-owner",
"email": "e2e.team-member-mail-owner@example.com",
"firstName": "E2E",
"lastName": "Group Member Mailbox Owner",
"enabled": true,
"credentials": [
{
"type": "password",
"value": "password-e2e.team-member-mail-owner"
}
],
"realmRoles": ["user"]
},
{
"username": "e2e.team-administrator-mail-member",
"email": "e2e.team-administrator-mail-member@example.com",
"firstName": "E2E",
"lastName": "Group Administrator Mailbox Member",
"enabled": true,
"credentials": [
{
"type": "password",
"value": "password-e2e.team-administrator-mail-member"
}
],
"realmRoles": ["user"]
},
{
"username": "e2e.team-administrator-mail-administrator",
"email": "e2e.team-administrator-mail-administrator@example.com",
"firstName": "E2E",
"lastName": "Group Administrator Mailbox Administrator",
"enabled": true,
"credentials": [
{
"type": "password",
"value": "password-e2e.team-administrator-mail-administrator"
}
],
"realmRoles": ["user"]
},
{
"username": "e2e.team-administrator-mail-owner",
"email": "e2e.team-administrator-mail-owner@example.com",
"firstName": "E2E",
"lastName": "Group Administrator Mailbox Owner",
"enabled": true,
"credentials": [
{
"type": "password",
"value": "password-e2e.team-administrator-mail-owner"
}
],
"realmRoles": ["user"]
},
{
"username": "e2e.team-owner-mail-member",
"email": "e2e.team-owner-mail-member@example.com",
"firstName": "E2E",
"lastName": "Group Owner Mailbox Member",
"enabled": true,
"credentials": [
{
"type": "password",
"value": "password-e2e.team-owner-mail-member"
}
],
"realmRoles": ["user"]
},
{
"username": "e2e.team-owner-mail-administrator",
"email": "e2e.team-owner-mail-administrator@example.com",
"firstName": "E2E",
"lastName": "Group Owner Mailbox Administrator",
"enabled": true,
"credentials": [
{
"type": "password",
"value": "password-e2e.team-owner-mail-administrator"
}
],
"realmRoles": ["user"]
},
{
"username": "e2e.team-owner-mail-owner",
"email": "e2e.team-owner-mail-owner@example.com",
"firstName": "E2E",
"lastName": "Mailbox Owner",
"enabled": true,
"credentials": [
{
"type": "password",
"value": "password-e2e.team-owner-mail-owner"
}
],
"realmRoles": ["user"]
}
],
"roles": {
@@ -712,17 +485,9 @@
"webAuthnPolicyPasswordlessAvoidSameAuthenticatorRegister": false,
"webAuthnPolicyPasswordlessAcceptableAaguids": [],
"scopeMappings": [
{
"clientScope": "siret",
"roles": [
"user"
]
},
{
"clientScope": "offline_access",
"roles": [
"offline_access"
]
"roles": ["offline_access"]
}
],
"clientScopeMappings": {
@@ -972,7 +737,6 @@
"acr",
"roles",
"profile",
"siret",
"email"
],
"optionalClientScopes": [
@@ -1133,35 +897,6 @@
}
]
},
{
"id": "eb220fbb-02ac-4105-95a3-727954f6565d",
"name": "siret",
"description": "siret",
"protocol": "openid-connect",
"attributes": {
"include.in.token.scope": "true",
"display.on.consent.screen": "false",
"gui.order": ""
},
"protocolMappers": [
{
"id": "333a4e89-9363-4c36-b56f-79c6b019c6c6",
"name": "siret",
"protocol": "openid-connect",
"protocolMapper": "oidc-usermodel-attribute-mapper",
"consentRequired": false,
"config": {
"aggregate.attrs": "false",
"userinfo.token.claim": "true",
"multivalued": "false",
"user.attribute": "siret",
"id.token.claim": "true",
"access.token.claim": "true",
"claim.name": "siret"
}
}
]
},
{
"id": "af52ccc3-4ecb-49b4-9a67-5d4172f16070",
"name": "role_list",
@@ -1628,8 +1363,7 @@
"email",
"roles",
"web-origins",
"acr",
"siret"
"acr"
],
"defaultOptionalClientScopes": [
"offline_access",
@@ -1,3 +1,37 @@
server {
listen 8082;
server_name localhost;
charset utf-8;
location /media {
alias /data/media;
}
location / {
proxy_pass http://app:8000;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
}
}
server {
listen 8088;
server_name localhost;
root /home/desk;
location / {
try_files $uri index.html $uri/ =404;
}
error_page 404 /404.html;
location = /404.html {
internal;
}
}
server {
listen 8083;
@@ -11,3 +45,4 @@ server {
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
}
}
-45
View File
@@ -1,45 +0,0 @@
# dimail
## What is dimail ?
The mailing solution provided in La Suite is [Open-XChange](https://www.open-xchange.com/) (OX).
OX not having a provisioning API, 'dimail-api' or 'dimail' was created to allow mail-provisioning through People.
API and its documentation can be found [here](https://api.dev.ox.numerique.gouv.fr/docs#/).
## Architectural links of dimail
As dimail's primary goal is to act as an interface between People and OX, its architecture is similar to that of People. A series of requests are sent from People to dimail upon creating domains, users, permissions and mailboxes.
### Domains
Upon creating a domain on People, the same domain is created on dimail and will undergo a series of checks. When all checks have passed, the domain is considered enabled.
Domains in OX have a field called "context". Context is a shared space between domains, allowing users to discover users not only on their domain but on their entire context.
### Mailboxes
Mailboxes can be created by a domain owners or administrators in People's domain tab.
On enabled domains, mailboxes are created at the same time on dimail (and a confirmation email is sent to the secondary email).
On pending/failed domains, mailboxes are only created locally with "pending" status and are sent to dimail upon domain's activation.
On disabled domains, mailboxes creation is not allowed.
### Users
The ones issuing requests. Dimail users will reflect domains owners and administrators on People, for logging purposes and to allow direct use of dimail, if desired. User reconciliation is made on user uuid provided by ProConnect.
### Permissions
As for People, an access - a permissions (or an "allow" in dimail) - grants an user permission to create objects on a domain.
Permissions requests are sent automatically upon :
- dimail database initialisation:
+ permission for dimail user People to create users and domains
- domain creation :
+ permission for dimail user People to manage domain
+ permission for new owner on new domain
- user creation:
+ permission for People to manage user
- access creation, if owner or admin:
+ permission for this user to manage domain
-131
View File
@@ -1,131 +0,0 @@
# Local development with Kubernetes
We use tilt to provide a local development environment for Kubernetes.
Tilt is a tool that helps you develop applications for Kubernetes.
It watches your files for changes, rebuilds your containers, and restarts your pods.
It's like having a conversation with your cluster.
## Prerequisites
This guide assumes you have the following tools installed:
- [Docker](https://docs.docker.com/get-docker/)
- [Kubectl](https://kubernetes.io/docs/tasks/tools/install-kubectl/)
- [Kind](https://kind.sigs.k8s.io/docs/user/quick-start/)
* [mkcert](https://github.com/FiloSottile/mkcert)
- [ctlptl](https://github.com/tilt-dev/ctlptl)
- [Tilt](https://docs.tilt.dev/install.html)
* [helm](https://helm.sh/docs/intro/install/)
* [helmfile](https://github.com/helmfile/helmfile)
* [secrets](https://github.com/jkroepke/helm-secrets/wiki/Installation)
* [sops](https://github.com/getsops/sops)
### SOPS configuration
**Generate a SOPS key**
For this specific step you need to have the `age-keygen` tool installed.
See https://github.com/FiloSottile/age.
Then generate a key:
```bash
age-keygen -o my-age.key
```
**Install the SOPS key**
Read the SOPS documentation on how to install the key in your environment.
https://github.com/getsops/sops?tab=readme-ov-file#22encrypting-using-age
On Ubuntu it's like:
```bash
mkdir -p ~/.config/sops/age/
cp my-age.key ~/.config/sops/age/keys.txt
chmod 400 ~/.config/sops/age/keys.txt
```
**Add the SOPS key to the repository**
Update the [.sops.yaml](../.sops.yaml) file with the **public** key id you generated.
### Helmfile in Docker
If you use helmfile in Docker, you may need an additional configuration to make
it work with you age key.
You need to mount `-v "${HOME}/.config/sops/age/:/helm/.config/sops/age/"`
```bash
#!/bin/sh
docker run --rm --net=host \
-v "${HOME}/.kube:/root/.kube" \
-v "${HOME}/.config/helm:/root/.config/helm" \
-v "${HOME}/.config/sops/age/:/helm/.config/sops/age/" \
-v "${HOME}/.minikube:/${HOME}/.minikube" \
-v "${PWD}:/wd" \
-e KUBECONFIG=/root/.kube/config \
--workdir /wd ghcr.io/helmfile/helmfile:v0.150.0 helmfile "$@"
```
## Getting started
### Create the kubernetes cluster
Run the following command to create a kubernetes cluster using kind:
```bash
./bin/start-kind.sh
```
**or** run the equivalent using the makefile
```bash
make start-kind
```
### [Optional] Install the secret
You don't need to do this if you are running the stack with keycloak.
```bash
make install-external-secrets
```
### Deploy the application
```bash
# Pro Connect environment
tilt up -f ./bin/Tiltfile
# Standalone environment with keycloak
DEV_ENV=dev-keycloak tilt up -f ./bin/Tiltfile
```
**or** run the equivalent using the makefile
```bash
# Pro Connect environment
make tilt-up
# Standalone environment with keycloak
make tilt-up-keycloak
```
That's it! You should now have a local development environment for Kubernetes.
You can access the application at https://desk.127.0.0.1.nip.io
## Management
To manage the cluster, you can use k9s.
## Next steps
- Add dimail to the local development environment
- Add a reset demo `cmd_button` to Tilt
+73 -103
View File
@@ -6,56 +6,16 @@ interact efficiently.
Let's see how we could interact with Django's shell to recreate David's environment in the app.
## Profile contact
## Base contacts from the organization records
The system identifies Dave through the `sub` associated with his OIDC session.
The OIDC should also provide his email address.
Note: as some Identity Providers do not provide a constant `sub`, the system should be able to
fallback to the email address to find it the user is already in the database.
When Dave logs in for the first time, the system checks if he is already in the database. If not
a new user is created for him.
Users needs to be linked to an Organization (see [organizations.md](./organizations.md)).
David Bowman is an exemplary employee at Space Odyssey Corporation. His email is
`david.bowman@spaceodyssey.com` and he is registered in the organization's records via a base
contact as follows:
```python
space_odyssey, _created = Organization.objects.get_or_create(...)
david_user = User.objects.create(
organization=space_odyssey,
language="en-us",
timezone="America/Los_Angeles",
name="David Bowman",
sub="1234567890",
email="david.bowman@spaceodyssey.com",
)
```
The system then creates a profile contact for Dave, based on the information provided by the OIDC:
```python
david_profile_contact = Contact.objects.create(
owner=david_user,
user=david_user, # this means it's the user's profile contact
david_base_contact = Contact.objects.create(
full_name="David Bowman",
data={
"emails": [
{"type": "Work", "value": "david.bowman@spaceodyssey.com"},
],
},
)
```
**Future feature:**
Dave will be prompted to confirm the details of the base contact stored in the database
and fill the details for a better profile.
His profile contact will be updated with the new information.
```python
Contact.objects.filter(user=david_user, owner=david_user).update(
short_name="Dave",
short_name="David",
data={
"emails": [
{"type": "Work", "value": "david.bowman@spaceodyssey.com"},
@@ -88,81 +48,91 @@ Contact.objects.filter(user=david_user, owner=david_user).update(
)
```
David's profile contact is available to all users of his company/organization.
When David logs-in to the People application for the first time using the corporation's OIDC
Single Sign-On service. A user is created for him on the fly by the system, together with an
identity record representing the OIDC session:
```python
david_user = User.objects.create(
language="en-us",
timezone="America/Los_Angeles",
)
david_identity = Identity.objects.create(
"user": david_user,
"sub": "2a1b3c4d-5e6f-7a8b-9c0d-1e2f3a4b5c6d",
"email" : "david.bowman@spaceodyssey.com",
"is_main": True,
)
```
## Profile contact
The system identifies Dave through the email associated with his OIDC session and prompts him to
confirm the details of the base contact stored in the database.
When David confirms, giving an alternative short name that he prefers, a contact override is
created on top of the organization's base contact. This new contact is marked as David's profile
on the user:
```python
david_contact = Contact.objects.create(
base=david_base_contact,
owner=david_user,
full_name="David Bowman",
short_name="Dave",
data={}
)
david_user.profile_contact = david_contact
david_user.save()
```
If Dave had not had any existing contact in the organization's records, the profile contact would
have been created independently, without any connection to a base contact:
```python
david_contact = Contact.objects.create(
base=None,
owner=david_user,
full_name="David Bowman",
short_name="Dave",
data={}
)
```
Now, Dave feels like sharing his mobile phone number with his colleagues. He can do this
by editing his contact in the application:
```python
contact.data["phones"] = [
{"type": "Mobile", "value": "(123) 456-7890"},
]
contact.save()
```
## Contact override
During a Space conference he attended, Dave met Dr Ryan Stone, a medical engineer who gave him
her personal email address (because she will quit the company she works for). Ryan is already
present in the system.
her professional email address. Ryan is already present in the system but her email is missing.
Dave can add it to his private version of the contact:
```python
ryan_user = User.objects.create(
organization=space_odyssey,
language="en-us",
timezone="America/Los_Angeles",
name="Ryan Stone",
sub="0987654321",
email="ryan.stone@spaceodyssey.com",
ryan_base_contact = Contact.objects.create(
full_name="Ryan Stone",
data={}
)
ryan_profile_contact = Contact.objects.create(
user=ryan_user,
owner=ryan_user,
ryan_contact = Contact.objects.create(
base=ryan_base_contact,
owner=david_user,
full_name="Ryan Stone",
short_name="Dr Ryan",
data={
"emails": [
{"type": "Work", "value": "ryan.stone@hubblestation.com"},
],
},
)
david_ryan_contact = Contact.objects.create(
override=ryan_profile_contact,
owner=david_user,
full_name="Ryan Stone",
short_name="Dr Ryan",
data={
"emails": [
{"type": "Home", "value": "ryan@stone.com"},
],
},
}
)
```
## Personal contacts
Dave met a lot of people during the last Opensource Experience event. He can add them to his personal contacts:
```python
julie_personal_contact = Contact.objects.create(
owner=david_user,
full_name="Julie Powell",
short_name="Julie",
data={
"emails": [
{"type": "Work", "value": "julie.powell@example.com"},
],
"phones": [
{"type": "Work", "value": "(123) 456-7890"},
],
},
)
```
These contacts are only available to Dave.
## Contacts from shared directories
This is a future feature that will allow Dave to access contacts from shared directories.
Shared directories will provide contacts for all users from several organizations.
## Team Collaboration
Dave wants to form a team with Ryan and other colleagues to work together better on using the organization's digital tools for their projects.
-41
View File
@@ -1,41 +0,0 @@
# Organization model
## Purpose
The initial `Organization` model allows to regroup `User` and `Team` under a same object.
While the purpose was purely technical in the first place, few features emerged afterward.
## Link with the `User` model
Each user must have a "default" organization.
When a user logs in:
- The backend tries to get an existing organization from a "registration ID" or from the user's mail domain
- If the organization does not exist, it is created using the "registration ID" and fallbacks on the user's mail domain
- The user organization is set to this organization.
**Note:** While deploying the new code, we allow the already existing user to not have one.
## Link with the `Team` model
Each team must have an organization. This organization is not defined by the user, but automatically set
using the user's organization.
The team's organization does not restrict the users who can be added to the team.
There is currently no feature relying on the team organization.
**Note:** While deploying the new code, we allow the teams to not have one.
## Organization-related features
### Organization UI
A new interface can be created to allow users to see all members of an organization.
This could be used along with the contacts.
-164
View File
@@ -1,164 +0,0 @@
## Resource Server
For detailed information, please refer to the [OAuth 2.0 Resource Server documentation](https://www.oauth.com/oauth2-servers/the-resource-server/) and review the relevant commits that implement the resource server backend.
---
#### Overview
A resource server is a crucial component in an OAuth 2.0 ecosystem. It is responsible for accepting access tokens issued by the authorization server (in this case, Agent Connect), introspecting and validating those tokens, and then returning the requested protected resources to the client.
By implementing a resource server, we can securely share data between different services within La Suite. This ensures that only clients with the appropriate scopes and permissions can access specific resources, thereby enhancing security and maintaining proper access control across services.
---
#### Disclaimer
- Currently compatible only with Agent Connect.
- The development setup requires simplification, with dependencies on Agent Connect ideally mocked.
- Terminology aligns with the specification: what is referred to as a "resource server" is known as a "data provider" in Agent Connect.
- This documentation is WIP.
---
## Running Locally
#### Prerequisites
- **Agent Connect Stack**: Ensure the local Agent Connect stack is running. A solid understanding of its configuration and operation is recommended (advanced level).
- **People Stack**: Make sure the People stack is up and running.
- **Ngrok**: Install and set up Ngrok for secure tunneling.
---
### Update People's configurations
#### Environment variables
Agent Connect includes two pre-configured mocked data providers in its default stack (`bdd-fca-low`).
Use the client ID and client secret from one of these data providers. **Note:** Make sure to retrieve the decrypted secret, as it is stored encrypted in the database. You can find these values in the `dp.js` fixture file, where they are exposed in a comment.
Configure your environment with the following values from Agent Connect:
```
OIDC_RS_CLIENT_ID=<your-client-id-from-ac>
OIDC_RS_CLIENT_SECRET=<your-decrypted-client-secret-from-ac>
# In development, the resource server use insecure settings
OIDC_VERIFY_SSL=False
# Update the endpoints as follows
OIDC_OP_JWKS_ENDPOINT=https://core-fca-low.docker.dev-franceconnect.fr/api/v2/jwks
OIDC_OP_INTROSPECTION_ENDPOINT=https://core-fca-low.docker.dev-franceconnect.fr/api/v2/checktoken
OIDC_OP_URL=https://core-fca-low.docker.dev-franceconnect.fr/api/v2
```
#### Docker Network Configuration
To enable communication between the Docker networks for People and Agent Connect, update your docker-compose configuration. This setup is required because the Authorization Server and Resource Server will exchange requests over a back-channel, necessitating their accessibility to each other.
1. **Create a Network**: Define a new network to bridge the two Docker networks.
```yaml
networks:
authorization_server:
external: true
driver: bridge
name: "${DESK_NETWORK:-fc_public}"
```
2. **Update Network for `app-dev`**: Ensure your `app-dev` service is connected to both the default network and the new `authorization_server` network.
```yaml
app-dev:
...
networks:
- default
- authorization_server
```
#### Ngrok
To expose your local resource server through an HTTP tunnel, use Ngrok. This is necessary because, in the Agent Connect development stack, the resource server needs to be accessible to a user agent via a publicly accessible URL.
```
$ ngrok http 8071
```
---
### Update AgentConnect's configurations
Modify the AgentConnect configuration to include the local resource server settings.
**Update `fsa1-low` Environment File**:
Add your Ngrok URL and configure the `DATA_APIS` list with the appropriate values.
```env
App_DATA_APIS=[{"name":"Data Provider 1","url":"https://your-ngrok-url/api/v1.0/any-path","secret":"***"}, ...]
```
**Update Fixture in `dp.js`**:
Adjust the configuration for the data provider to match your local setup. Ensure that the `client_id`, `client_secret`, and other parameters are correctly set and aligned with your environment. This can be configured through environment variables.
```javascript
const dps = [
// Data Provider Configuration
{
uid: "6f21b751-ed06-48b6-a59c-36e1300a368a",
title: "Mock Data Provider - 1",
active: true,
slug: "DESK",
client_id: "***",
client_secret: "***",
// client_secret decrypted : ****
jwks_uri: "https://your-ngrok-url/api/v1.0/jwks", // Update this line
checktoken_signed_response_alg: "ES256",
checktoken_encrypted_response_alg: "RSA-OAEP",
checktoken_encrypted_response_enc: "A256GCM",
},
];
```
**Note**: Ensure that the `jwks_uri` and other cryptographic parameters (e.g., `checktoken_signed_response_alg`, `checktoken_encrypted_response_alg`, and `checktoken_encrypted_response_enc`) match your actual setup and are configured via environment variables where necessary.
---
### Usage
This section is a work in progress. Please note the following important points:
#### User `sub` Matching
Ensure that the `sub` (subject) field for users in AgentConnect matches the corresponding value in the People database. To synchronize this, you can run `make demo`, then edit the user's `sub` field to match the value returned by AgentConnect. For this, you'll need to update the editable field in Django Admin, specifically in `admin.py`. Adjust the `get_readonly_fields` method as follows:
```python
def get_readonly_fields(self, request, obj=None):
"""The 'sub' field should only be editable during creation, not for updates."""
if obj:
return self.readonly_fields
return self.readonly_fields + ["sub"] # update this line adding 'sub'
```
#### Scope for `groups`
Ensure that the `groups` scope is requested from the service provider during authentication with AgentConnect.
#### Resource Server Requests
By default, the `fsa1-low` environment calls the resource server using a POST request.
#### Testing
Most of the testing has been done using the `/users/me` endpoint. Update the `api/viewset.py` configuration to allow both GET and POST methods for this endpoint:
```python
@decorators.action(
detail=False,
methods=["get", "post"], # update this line adding 'post'
url_name="me",
url_path="me",
)
```
-28
View File
@@ -1,28 +0,0 @@
# ServiceProvider model
## Purpose
The `ServiceProvider` model represents a ... service provider, also known as "tools using some data from this project".
## Link with the `Organization` model
An organization can be linked to several service providers.
The goal here, is to allow users of an organization to have access, to a service provider or not.
The first implementation does not have any feature related, but the first feature will probably be
to list applications visible in the "all application menu" (aka "la gaufre").
## Link with the `Team` model
A team can be linked to several service providers.
This is used as a filter when a service provider calls the resource server, only the teams linked to
this service provider are returned. This is mandatory for data segregation: we don't want all service
providers to be able to list all data regarding other service providers.
## Limitations
There is currently no way to provision all the service providers automatically. So when a service provider
creates a team via the resource server, we create the `ServiceProvider` on the fly, without any understandable name.
This will need to be improved later.
-51
View File
@@ -1,51 +0,0 @@
# E2E tests
## Run E2E tests
``` bash
# you need the dockers to be up and running
make bootstrap
# you will need to have few accounts in the database
make demo FLUSH_ARGS='--no-input'
# run the tests
cd src/frontend/apps/e2e
yarn test:ui --workers=1
```
A new browser window will open and you will be able to run the tests.
## Available accounts
The `make demo` command creates the following accounts:
- `e2e.team-<role>@example.com` where `<role>` is one of `administrator`, `member`, `owner`:
this account only belong to a team with the specified role.
- `e2e.mail-<role>@example.com` where `<role>` is one of `administrator`, `member`, `owner`:
this account only have a mailbox with the specified role access.
- `e2e.team-<team_role>-mail-<domain_role>@example.com` with a combination of roles as for the
previous accounts.
For each account, the password is `password-e2e.<role>`, for instance `password-e2e.team-member`.
In the E2E tests you can use these accounts to benefit from there accesses,
using the `keyCloakSignIn(page, browserName, <account_name>)`. The account name is the
username without the prefix `e2e.`.
``` typescript jsx
await keyCloakSignIn(page, browserName, 'mail-owner');
```
The `keyCloakSignIn` function will sign in the user on Keycloak using the proper username and password.
.. note::
This only works because the OIDC setting is set to fallback on user email.
## Add a new account
In case you need to add a new account for specific tests you need:
- to create a new user with the same format in the backend database:
update `[create_demo.py](../src/backend/demo/management/commands/create_demo.py)`
- to create a new account in Keycloak: update [realm.json](../docker/auth/realm.json)
- if the keycloak was running locally, you need to destroy its database and
restart the database and the keycloak containers.
-32
View File
@@ -33,35 +33,3 @@ LOGOUT_REDIRECT_URL=http://localhost:3000
OIDC_REDIRECT_ALLOWED_HOSTS=["http://localhost:8083", "http://localhost:3000"]
OIDC_AUTH_REQUEST_EXTRA_PARAMS={"acr_values": "eidas1"}
OIDC_RS_CLIENT_ID=people
OIDC_RS_CLIENT_SECRET=ThisIsAnExampleKeyForDevPurposeOnly
OIDC_RS_PRIVATE_KEY_STR="-----BEGIN PRIVATE KEY-----
MIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQC3boG1kwEGUYL+
U58RPrVToIsF9jHB64S6WJIIInPmAclBciXFb6BWG11mbRIgo8ha3WVnC/tGHbXb
ndiKdrH2vKHOsDhV9AmgHgNgWaUK9L0uuKEb/xMLePYWsYlgzcQJx8RZY7RQyWqE
20WfzFxeuCE7QMb6VXSOgwQMnJsKocguIh3VCI9RIBq3B1kdgW35AD63YKOygmGx
qjcWwbjhKLvkF7LpBdlyAEzOKqg4T5uCcHMfksMW2+foTJx70RrZM/KHU+Zysuw7
uhhVsgPBG+CsqBSjHQhs7jzymqxtQAfe1FkrCRxOq5Pv2Efr7kgtVSkJJiX3KutM
vnWuEypxAgMBAAECggEAGqKS9pbrN+vnmb7yMsqYgVVnQn0aggZNHlLkl4ZLLnuV
aemlhur7zO0JzajqUC+AFQOfaQxiFu8S/FoJ+qccFdATrcPEVmTKbgPVqSyzLKlX
fByGll5eOVT95NMwN8yBGgt2HSW/ZditXS/KxxahVgamGqjAC9MTSutGz/8Ae1U+
DNDBJCc6RAqu3T02tV9A2pSpVC1rSktDMpLUTscnsfxpaEQATd9DJUcHEvIwoX8q
GJpycPEhNhdPXqpln5SoMHcf/zS5ssF/Mce0lJJXYyE0LnEk9X12jMWyBqmLqXUY
cKLyynaFbis0DpQppwKx2y8GpL76k+Ci4dOHIvFknQKBgQDj/2WRMcWOvfBrggzj
FHpcme2gSo5A5c0CVyI+Xkf1Zab6UR6T7GiImEoj9tq0+o2WEix9rwoypgMBq8rz
/rrJAPSZjgv6z71k4EnO2FIB5R03vQmoBRCN8VlgvLM0xv52zyjV4Wx66Q4MDjyH
EgkpHyB0FzRZh0UzhnE/pYSetQKBgQDN9eLB1nA4CBSr1vMGNfQyfBQl3vpO9EP4
VSS3KnUqCIjJeLu682Ylu7SFxcJAfzUpy5S43hEvcuJsagsVKfmCAGcYZs9/xq3I
vzYyhaEOS5ezNxLSh4+yCNBPlmrmDyoazag0t8H8YQFBN6BVcxbATHqdWGUhIhYN
eEpEMOh2TQKBgGBr7kRNTENlyHtu8IxIaMcowfn8DdUcWmsW9oBx1vTNHKTYEZp1
bG/4F8LF7xCCtcY1wWMV17Y7xyG5yYcOv2eqY8dc72wO1wYGZLB5g5URlB2ycJcC
LVIaM7ZZl2BGl+8fBSIOx5XjYfFvQ+HLmtwtMchm19jVAEseHF7SXRfRAoGAK15j
aT2mU6Yf9C9G7T/fM+I8u9zACHAW/+ut14PxN/CkHQh3P16RW9CyqpiB1uLyZuKf
Zm4cYElotDuAKey0xVMgYlsDxnwni+X3m5vX1hLE1s/5/qrc7zg75QZfbCI1U3+K
s88d4e7rPLhh4pxhZgy0pP1ADkIHMr7ppIJH8OECgYEApNfbgsJVPAMzucUhJoJZ
OmZHbyCtJvs4b+zxnmhmSbopifNCgS4zjXH9qC7tsUph1WE6L2KXvtApHGD5H4GQ
IH5em4M/pHIcsqCi1qggBMbdvzHBUtC3R4sK0CpEFHlN+Y59aGazidcN2FPupNJv
MbyqKyC6DAzv4jEEhHaN7oY=
-----END PRIVATE KEY-----
"
+1 -1
View File
@@ -8,4 +8,4 @@ DB_HOST=kc_postgresql
DB_NAME=keycloak
DB_USER=people
DB_PASSWORD=pass
DB_PORT=5433
DB_PORT=5433
+1 -1
View File
@@ -8,4 +8,4 @@ DB_HOST=postgresql
DB_NAME=people
DB_USER=dinum
DB_PASSWORD=pass
DB_PORT=5432
DB_PORT=5432
+15 -2
View File
@@ -1,7 +1,6 @@
{
"extends": ["github>numerique-gouv/renovate-configuration"],
"dependencyDashboard": true,
"labels": ["dependencies", "noChangeLog"],
"packageRules": [
{
"enabled": false,
@@ -13,7 +12,21 @@
"enabled": false,
"groupName": "ignored js dependencies",
"matchManagers": ["npm"],
"matchPackageNames": ["fetch-mock", "node", "node-fetch", "eslint"]
"matchPackageNames": ["node", "node-fetch", "i18next-parser"]
},
{
"enabled": false,
"groupName": "ignored js @types/react-dom@18.2.19",
"matchManagers": ["npm"],
"matchPackageNames": ["@types/react-dom"],
"allowedVersions": ">18.2.19"
},
{
"enabled": false,
"groupName": "ignored js @openfun/cunningham-react@2.5.0",
"matchManagers": ["npm"],
"matchPackageNames": ["@openfun/cunningham-react"],
"allowedVersions": ">2.5.0"
}
]
}
+2 -2
View File
@@ -1,10 +1,10 @@
#!/usr/bin/env bash
#!/bin/bash
mkdir -p "$(dirname -- "${BASH_SOURCE[0]}")/../.git/hooks/"
PRE_COMMIT_FILE="$(dirname -- "${BASH_SOURCE[0]}")/../.git/hooks/pre-commit"
cat <<'EOF' >$PRE_COMMIT_FILE
#!/usr/bin/env bash
#!/bin/bash
# directories containing potential secrets
DIRS="."
-161
View File
@@ -1,161 +0,0 @@
# pylint: disable=line-too-long
import datetime
import os
import re
import sys
from utils import run_command
RELEASE_KINDS = {'p': 'patch', 'm': 'minor', 'mj': 'major'}
def update_files(version):
"""Update all files needed with new release version"""
# pyproject.toml
sys.stdout.write("Update pyproject.toml...\n")
path = "src/backend/pyproject.toml"
with open(path, 'r') as file:
lines = file.readlines()
for index, line in enumerate(lines):
if line.startswith("version = "):
lines[index] = re.sub(r'\"(.*?)\"', f'"{version}"', line)
with open(path, 'w+') as file:
file.writelines(lines)
# frontend package.json
sys.stdout.write("Update package.json...\n")
files_to_modify = []
filename = "package.json"
for root, _dir, files in os.walk("src/frontend"):
if filename in files and "node_modules" not in root and ".next" not in root:
files_to_modify.append(os.path.join(root, filename))
for path in files_to_modify:
with open(path, 'r') as file:
lines = file.readlines()
for index, line in enumerate(lines):
if "version" in line:
lines[index] = re.sub(r'"version": \"(.*?)\"', f'"version": "{version}"', line)
with open(path, 'w+') as file:
file.writelines(lines)
return
def update_helm_files(path):
sys.stdout.write("Update helm files...\n")
with open(path, 'r') as file:
lines = file.readlines()
for index, line in enumerate(lines):
if "tag:" in line:
lines[index] = re.sub(r'\"(.*?)\"', f'"v{version}"', line)
with open(path, 'w+') as file:
file.writelines(lines)
def update_changelog(path, version):
"""Update changelog file with release info
"""
sys.stdout.write("Update CHANGELOG...\n")
with open(path, 'r') as file:
lines = file.readlines()
for index, line in enumerate(lines):
if "## [Unreleased]" in line:
today = datetime.date.today()
lines.insert(index + 1, f"\n## [{version}] - {today}\n")
if line.startswith("[unreleased]"):
last_version = lines[index + 1].split("]")[0][1:]
new_unreleased_line = line.replace(last_version, version)
new_release_line = lines[index + 1].replace(last_version, version)
lines[index] = new_unreleased_line
lines.insert(index + 1, new_release_line)
break
with open(path, 'w+') as file:
file.writelines(lines)
def deployment_part(version, kind):
""" Update helm file of preprod on deployment repository numerique-gouv/lasuite-deploiement
"""
# Move to lasuite-deploiement repository
try:
os.chdir('../lasuite-deploiement')
except FileNotFoundError:
sys.stdout.write("\033[1;31m[Error] You must have a clone of https://github.com/numerique-gouv/lasuite-deploiement\x1b[0m")
sys.stdout.write("\033[0;32mPlease clone it and re-run script with option --only-deployment-part\x1b[0m")
sys.stdout.write(" >>> python scripts/release.py --only-deployment-part \x1b[0m")
else:
run_command("git checkout main", shell=True)
run_command("git pull", shell=True)
deployment_branch = f"regie/{version}/preprod"
run_command(f"git checkout -b {deployment_branch}", shell=True)
run_command("git pull --rebase origin main", shell=True)
path = f"manifests/regie/env.d/preprod/values.desk.yaml.gotmpl"
update_helm_files(path)
run_command(f"git add {path}", shell=True)
message = f"""🔖({RELEASE_KINDS[kind]}) release version {version}"""
run_command(f"git commit -m '{message}'", shell=True)
confirm = input(f"""\033[0;32m
### DEPLOYMENT ###
NEXT COMMAND on lasuite-deploiement repository:
>> git push origin {deployment_branch}
Continue ? (y,n)
\x1b[0m""")
if confirm == 'y':
run_command(f"git push origin {deployment_branch}", shell=True)
sys.stdout.write(f"""\033[1;34m
PLEASE DO THE FOLLOWING INSTRUCTIONS:
--> Please submit PR {deployment_branch} and merge code to main
\x1b[0m""")
def project_part(version, kind):
"""Manage only la regie part with update of CHANGELOG, version files and tag"""
sys.stdout.write('Let\'s go to create branch to release\n')
branch_to_release = f"release/{version}"
run_command(f"git checkout -b {branch_to_release}", shell=True)
run_command("git pull --rebase origin main", shell=True)
update_changelog("CHANGELOG.md", version)
update_files(version)
run_command("git add CHANGELOG.md", shell=True)
run_command("git add src/", shell=True)
message = f"""🔖({RELEASE_KINDS[kind]}) release version {version}
Update all version files and changelog for {RELEASE_KINDS[kind]} release."""
run_command(f"git commit -m '{message}'", shell=True)
confirm = input(f"""\033[0;32m
### RELEASE ###
NEXT COMMAND on current repository:
>> git push origin {branch_to_release}
Continue ? (y,n)
\x1b[0m""")
if confirm == 'y':
run_command(f"git push origin {branch_to_release}", shell=True)
sys.stdout.write(f"""
\033[1;34mPLEASE DO THE FOLLOWING INSTRUCTIONS:
--> Please submit PR {branch_to_release} and merge code to main
--> Then do:
>> git checkout main
>> git pull
>> git tag v{version}
>> git push origin v{version}
--> Please check and wait for the docker image v{version} to be published here:
- https://hub.docker.com/r/lasuite/people-backend/tags
- https://hub.docker.com/r/lasuite/people-frontend/tags
--> Now please generate release on github interface for the current tag v{version}
\x1b[0m""")
if __name__ == "__main__":
version, kind = None, None
while not version:
version = input("Enter your release version:")
while kind not in RELEASE_KINDS:
kind = input("Enter kind of release (p:patch, m:minor, mj:major):")
if "--only-deployment-part" not in sys.argv:
project_part(version, kind)
deployment_part(version, kind)
-4
View File
@@ -1,4 +0,0 @@
#!/bin/bash
git submodule update --init --recursive
git submodule foreach 'git fetch origin; git checkout $(git rev-parse --abbrev-ref HEAD); git reset --hard origin/$(git rev-parse --abbrev-ref HEAD); git submodule update --recursive; git clean -dfx'
-14
View File
@@ -1,14 +0,0 @@
import subprocess
import sys
def run_command(cmd, msg=None, shell=False, cwd='.', stdout=None):
if msg is None:
msg = f"# Running: {cmd}"
if stdout is not None:
stdout.write(msg + '\n')
if stdout != sys.stdout:
sys.stdout(msg)
subprocess.check_call(cmd, shell=shell, cwd=cwd, stdout=stdout, stderr=stdout)
if stdout is not None:
stdout.flush()
Submodule secrets deleted from b7ab5f1411
+1 -2
View File
@@ -363,8 +363,7 @@ name-group=
# Regular expression which should only match function or class names that do
# not require a docstring.
# Ignore: private stuff and `Params` class from FactoryBoy
no-docstring-rgx=(^_|^Params$)
no-docstring-rgx=^_
# List of decorators that produce properties, such as abc.abstractproperty. Add
# to this list to register other decorators that produce valid properties.
-1
View File
@@ -1 +0,0 @@
"""Root module."""
-1
View File
@@ -1 +0,0 @@
"""People custom admin site."""
-9
View File
@@ -1,9 +0,0 @@
"""Custom Django admin site application configuration."""
from django.contrib.admin.apps import AdminConfig
class PeopleAdminConfig(AdminConfig):
"""Declare our custom Django admin site."""
default_site = "admin.sites.PeopleAdminSite"
-15
View File
@@ -1,15 +0,0 @@
"""Custom Django admin site for the People app."""
from django.conf import settings
from django.contrib import admin
class PeopleAdminSite(admin.AdminSite):
"""People custom admin site."""
def each_context(self, request):
"""Add custom context to the admin site."""
return super().each_context(request) | {
"ADMIN_HEADER_BACKGROUND": settings.ADMIN_HEADER_BACKGROUND,
"ADMIN_HEADER_COLOR": settings.ADMIN_HEADER_COLOR,
}
@@ -1,13 +0,0 @@
{% extends "admin/base_site.html" %}
{% block extrastyle %}{{ block.super }}
<style>
html[data-theme="light"], :root {
{% if ADMIN_HEADER_BACKGROUND %}--header-bg: {{ ADMIN_HEADER_BACKGROUND }};{% endif %}
{% if ADMIN_HEADER_COLOR %}--header-color: {{ ADMIN_HEADER_COLOR }};{% endif %}
}
ul.messagelist li.info {
background-color: #EEEEEE;
}
</style>
{% endblock %}
-1
View File
@@ -1 +0,0 @@
"""Core root module."""
+47 -175
View File
@@ -1,19 +1,46 @@
"""Admin classes and registrations for People's core app."""
from django.contrib import admin, messages
from django import forms
from django.contrib import admin
from django.contrib.auth import admin as auth_admin
from django.utils.translation import gettext_lazy as _
from treebeard.admin import TreeAdmin
from treebeard.forms import movenodeform_factory
from mailbox_manager.admin import MailDomainAccessInline
from . import models
from .plugins.loader import (
get_organization_plugins,
organization_plugins_run_after_create,
)
class IdentityFormSet(forms.BaseInlineFormSet):
"""
Make the "is_main" field readonly when it is True so that declaring another identity
works in the admin.
"""
def add_fields(self, form, index):
"""Disable the "is_main" field when it is set to True"""
super().add_fields(form, index)
is_main_value = form.instance.is_main if form.instance else False
form.fields["is_main"].disabled = is_main_value
class IdentityInline(admin.TabularInline):
"""Inline admin class for user identities."""
fields = (
"sub",
"email",
"is_main",
"created_at",
"updated_at",
)
formset = IdentityFormSet
model = models.Identity
extra = 0
readonly_fields = ("email", "created_at", "sub", "updated_at")
def has_add_permission(self, request, obj):
"""
Identities are automatically created on successful OIDC logins.
Disable creating identities via the admin.
"""
return False
class TeamAccessInline(admin.TabularInline):
@@ -25,42 +52,21 @@ class TeamAccessInline(admin.TabularInline):
readonly_fields = ("created_at", "updated_at")
class OrganizationAccessInline(admin.TabularInline):
"""Inline admin class for organization accesses."""
autocomplete_fields = ["user", "organization"]
extra = 0
model = models.OrganizationAccess
readonly_fields = ("created_at", "updated_at")
class TeamWebhookInline(admin.TabularInline):
"""Inline admin class for team webhooks."""
extra = 0
autocomplete_fields = ["team"]
model = models.TeamWebhook
readonly_fields = ("created_at", "updated_at")
@admin.register(models.User)
class UserAdmin(auth_admin.UserAdmin):
"""Admin class for the User model"""
autocomplete_fields = ["organization"]
fieldsets = (
(
None,
{
"fields": (
"id",
"sub",
"password",
"organization",
)
},
),
(_("Personal info"), {"fields": ("name", "email", "language", "timezone")}),
(_("Personal info"), {"fields": ("email", "language", "timezone")}),
(
_("Permissions"),
{
@@ -81,14 +87,13 @@ class UserAdmin(auth_admin.UserAdmin):
None,
{
"classes": ("wide",),
"fields": ("sub", "email", "password1", "password2"),
"fields": ("email", "password1", "password2"),
},
),
)
inlines = (TeamAccessInline, MailDomainAccessInline, OrganizationAccessInline)
inlines = (IdentityInline, TeamAccessInline)
list_display = (
"get_user",
"organization",
"email",
"created_at",
"updated_at",
"is_active",
@@ -98,59 +103,22 @@ class UserAdmin(auth_admin.UserAdmin):
)
list_filter = ("is_staff", "is_superuser", "is_device", "is_active")
ordering = ("is_active", "-is_superuser", "-is_staff", "-is_device", "-updated_at")
readonly_fields = ["id", "created_at", "updated_at"]
search_fields = ("id", "email", "sub", "name")
def get_readonly_fields(self, request, obj=None):
"""The sub should only be editable for a create, not for updates."""
if obj:
return self.readonly_fields + ["sub"]
return self.readonly_fields
def get_user(self, obj):
"""Provide a nice display for user"""
return (
obj.name if obj.name else (obj.email if obj.email else f"[sub] {obj.sub}")
)
get_user.short_description = _("User")
class TeamServiceProviderInline(admin.TabularInline):
"""Inline admin class for service providers."""
can_delete = False
model = models.Team.service_providers.through
extra = 0
readonly_fields = ("id", "created_at", "updated_at")
search_fields = ("id", "email", "identities__sub", "identities__email")
@admin.register(models.Team)
class TeamAdmin(TreeAdmin):
class TeamAdmin(admin.ModelAdmin):
"""Team admin interface declaration."""
form = movenodeform_factory(models.Team)
inlines = (TeamAccessInline, TeamWebhookInline, TeamServiceProviderInline)
exclude = ("service_providers",) # Handled by the inline
inlines = (TeamAccessInline,)
list_display = (
"name",
"slug",
"created_at",
"updated_at",
)
search_fields = ("name",)
readonly_fields = ("path", "depth", "numchild")
@admin.register(models.TeamAccess)
class TeamAccessAdmin(admin.ModelAdmin):
"""Team access admin interface declaration."""
list_display = (
"user",
"team",
"role",
"created_at",
"updated_at",
)
@admin.register(models.Invitation)
@@ -176,102 +144,6 @@ class InvitationAdmin(admin.ModelAdmin):
"is_expired",
)
def get_readonly_fields(self, request, obj=None):
"""Mark all fields read only, i.e. disable update."""
if obj:
return self.fields
return self.readonly_fields
def change_view(self, request, object_id, form_url="", extra_context=None):
"""Custom edit form. Remove 'save' buttons."""
extra_context = extra_context or {}
extra_context["show_save_and_continue"] = False
extra_context["show_save"] = False
extra_context["show_save_and_add_another"] = False
return super().change_view(request, object_id, extra_context=extra_context)
def save_model(self, request, obj, form, change):
"""Fill in current logged-in user as issuer."""
obj.issuer = request.user
obj.save()
@admin.register(models.Contact)
class ContactAdmin(admin.ModelAdmin):
"""Contact admin interface declaration."""
list_display = (
"full_name",
"owner",
"override",
)
class OrganizationServiceProviderInline(admin.TabularInline):
"""Inline admin class for service providers."""
can_delete = False
model = models.Organization.service_providers.through
extra = 0
@admin.action(description=_("Run post creation plugins"), permissions=["change"])
def run_post_creation_plugins(modeladmin, request, queryset): # pylint: disable=unused-argument
"""Run the post creation plugins for the selected organizations."""
for organization in queryset:
organization_plugins_run_after_create(organization)
messages.success(
request,
_("Post creation plugins have been run for the selected organizations."),
)
@admin.register(models.Organization)
class OrganizationAdmin(admin.ModelAdmin):
"""Admin interface for organizations."""
actions = [run_post_creation_plugins]
list_display = (
"name",
"created_at",
"updated_at",
)
search_fields = ("name",)
inlines = (OrganizationAccessInline, OrganizationServiceProviderInline)
exclude = ("service_providers",) # Handled by the inline
def get_actions(self, request):
"""Adapt actions list to the context."""
actions = super().get_actions(request)
if not get_organization_plugins():
actions.pop("run_post_creation_plugins", None)
return actions
@admin.register(models.OrganizationAccess)
class OrganizationAccessAdmin(admin.ModelAdmin):
"""Organization access admin interface declaration."""
autocomplete_fields = ("user", "organization")
list_display = (
"user",
"organization",
"role",
"created_at",
"updated_at",
)
@admin.register(models.ServiceProvider)
class ServiceProviderAdmin(admin.ModelAdmin):
"""Admin interface for service providers."""
list_display = (
"name",
"audience_id",
"created_at",
"updated_at",
)
search_fields = ("name", "audience_id")
readonly_fields = ("created_at", "updated_at")
-3
View File
@@ -1,5 +1,4 @@
"""People core API endpoints"""
from django.conf import settings
from django.core.exceptions import ValidationError
@@ -23,8 +22,6 @@ def exception_handler(exc, context):
detail = exc.message
elif hasattr(exc, "messages"):
detail = exc.messages
else:
detail = ""
exc = drf_exceptions.ValidationError(detail=detail)
-1
View File
@@ -1 +0,0 @@
"""People core client API endpoints"""
-661
View File
@@ -1,661 +0,0 @@
"""API endpoints"""
import datetime
import operator
from functools import reduce
from django.conf import settings
from django.db.models import OuterRef, Q, Subquery, Value
from django.db.models.functions import Coalesce
from django.utils.decorators import method_decorator
from django.views.decorators.cache import cache_page
from rest_framework import (
decorators,
exceptions,
filters,
mixins,
pagination,
response,
throttling,
views,
viewsets,
)
from rest_framework.permissions import AllowAny
from core import models
from core.api import permissions
from core.api.client import serializers
from core.utils.raw_sql import gen_sql_filter_json_array
from mailbox_manager import models as domains_models
class NestedGenericViewSet(viewsets.GenericViewSet):
"""
A generic Viewset aims to be used in a nested route context.
e.g: `/api/v1.0/resource_1/<resource_1_pk>/resource_2/<resource_2_pk>/`
It allows to define all url kwargs and lookup fields to perform the lookup.
"""
lookup_fields: list[str] = ["pk"]
lookup_url_kwargs: list[str] = []
def __getattribute__(self, item):
"""
This method is overridden to allow to get the last lookup field or lookup url kwarg
when accessing the `lookup_field` or `lookup_url_kwarg` attribute. This is useful
to keep compatibility with all methods used by the parent class `GenericViewSet`.
"""
if item in ["lookup_field", "lookup_url_kwarg"]:
return getattr(self, item + "s", [None])[-1]
return super().__getattribute__(item)
def get_queryset(self):
"""
Get the list of items for this view.
`lookup_fields` attribute is enumerated here to perform the nested lookup.
"""
queryset = super().get_queryset()
# The last lookup field is removed to perform the nested lookup as it corresponds
# to the object pk, it is used within get_object method.
lookup_url_kwargs = (
self.lookup_url_kwargs[:-1]
if self.lookup_url_kwargs
else self.lookup_fields[:-1]
)
filter_kwargs = {}
for index, lookup_url_kwarg in enumerate(lookup_url_kwargs):
if lookup_url_kwarg not in self.kwargs:
raise KeyError(
f"Expected view {self.__class__.__name__} to be called with a URL "
f'keyword argument named "{lookup_url_kwarg}". Fix your URL conf, or '
"set the `.lookup_fields` attribute on the view correctly."
)
filter_kwargs.update(
{self.lookup_fields[index]: self.kwargs[lookup_url_kwarg]}
)
return queryset.filter(**filter_kwargs)
class SerializerPerActionMixin:
"""
A mixin to allow to define serializer classes for each action.
This mixin is useful to avoid to define a serializer class for each action in the
`get_serializer_class` method.
Example:
```
class MyViewSet(SerializerPerActionMixin, viewsets.GenericViewSet):
serializer_class = MySerializer
list_serializer_class = MyListSerializer
retrieve_serializer_class = MyRetrieveSerializer
```
"""
def get_serializer_class(self):
"""
Return the serializer class to use depending on the action.
"""
if serializer_class := getattr(self, f"{self.action}_serializer_class", None):
return serializer_class
return super().get_serializer_class()
class Pagination(pagination.PageNumberPagination):
"""Pagination to display no more than 100 objects per page sorted by creation date."""
max_page_size = 100
page_size_query_param = "page_size"
class BurstRateThrottle(throttling.UserRateThrottle):
"""
Throttle rate for minutes. See DRF section in settings for default value.
"""
scope = "burst"
class SustainedRateThrottle(throttling.UserRateThrottle):
"""
Throttle rate for hours. See DRF section in settings for default value.
"""
scope = "sustained"
# pylint: disable=too-many-ancestors
class ContactViewSet(
mixins.CreateModelMixin,
mixins.DestroyModelMixin,
mixins.RetrieveModelMixin,
mixins.UpdateModelMixin,
viewsets.GenericViewSet,
):
"""Contact ViewSet"""
permission_classes = [permissions.AccessPermission]
queryset = models.Contact.objects.select_related("user").all()
serializer_class = serializers.ContactSerializer
throttle_classes = [BurstRateThrottle, SustainedRateThrottle]
ordering_fields = ["full_name", "short_name", "created_at"]
ordering = ["full_name"]
def list(self, request, *args, **kwargs):
"""Limit listed users by a query with throttle protection."""
user = self.request.user
queryset = self.filter_queryset(self.get_queryset())
# List only contacts that:
queryset = queryset.filter(
# - is public (owner is None)
Q(owner__isnull=True)
# - are owned by the user
| Q(owner=user)
# - are profile contacts for a user from the same organization
| Q(user__organization_id=user.organization_id),
# - are not overriden by another contact
overridden_by__isnull=True,
)
# Search by case-insensitive and accent-insensitive on:
# - full name
# - short name
# - email (from data `emails` field)
if query := self.request.GET.get("q", ""):
queryset = queryset.filter(
Q(full_name__unaccent__icontains=query)
| Q(short_name__unaccent__icontains=query)
| Q(
id__in=gen_sql_filter_json_array(
queryset.model,
"data->'emails'",
"value",
query,
)
)
)
serializer = self.get_serializer(queryset, many=True)
return response.Response(serializer.data)
def perform_create(self, serializer):
"""Set the current user as owner of the newly created contact."""
user = self.request.user
serializer.validated_data["owner"] = user
return super().perform_create(serializer)
class OrganizationViewSet(
mixins.RetrieveModelMixin,
mixins.UpdateModelMixin,
viewsets.GenericViewSet,
):
"""
Organization ViewSet
GET /api/organizations/<organization_id>/
Return the organization details for the given id.
PUT /api/organizations/<organization_id>/
Update the organization details for the given id.
PATCH /api/organizations/<organization_id>/
Partially update the organization details for the given id.
"""
permission_classes = [permissions.AccessPermission]
queryset = models.Organization.objects.all()
serializer_class = serializers.OrganizationSerializer
throttle_classes = [BurstRateThrottle, SustainedRateThrottle]
def get_queryset(self):
"""Limit listed organizations to the one the user belongs to."""
return (
super()
.get_queryset()
.filter(pk=self.request.user.organization_id)
.annotate(
user_role=Subquery(
models.OrganizationAccess.objects.filter(
user=self.request.user, organization=OuterRef("pk")
).values("role")[:1]
)
)
)
class UserViewSet(
SerializerPerActionMixin,
mixins.UpdateModelMixin,
viewsets.GenericViewSet,
mixins.ListModelMixin,
):
"""
User viewset for all interactions with user infos and teams.
GET /api/users/&q=query
Return a list of users whose email or name matches the query.
"""
permission_classes = [permissions.IsSelf]
queryset = (
models.User.objects.select_related("organization").all().order_by("-created_at")
)
serializer_class = serializers.UserSerializer
get_me_serializer_class = serializers.UserMeSerializer
throttle_classes = [BurstRateThrottle, SustainedRateThrottle]
pagination_class = Pagination
def get_queryset(self):
"""Limit listed users by a query. Pagination and throttle protection apply."""
queryset = self.queryset
if self.action == "list":
# Exclude inactive contacts
queryset = queryset.filter(
is_active=True,
)
# Exclude all users already in the given team
if team_id := self.request.GET.get("team_id", ""):
queryset = queryset.exclude(teams__id=team_id)
# Search by case-insensitive and accent-insensitive
if query := self.request.GET.get("q", ""):
queryset = queryset.filter(
Q(name__unaccent__icontains=query)
| Q(email__unaccent__icontains=query)
)
return queryset
@decorators.action(
detail=False,
methods=["get"],
url_name="me",
url_path="me",
)
def get_me(self, request):
"""
Return information on currently logged user
"""
user = request.user
return response.Response(self.get_serializer(user).data)
class TeamViewSet(
mixins.CreateModelMixin,
mixins.DestroyModelMixin,
mixins.ListModelMixin,
mixins.RetrieveModelMixin,
mixins.UpdateModelMixin,
viewsets.GenericViewSet,
):
"""Team ViewSet"""
permission_classes = [permissions.TeamPermission, permissions.AccessPermission]
serializer_class = serializers.TeamSerializer
filter_backends = [filters.OrderingFilter]
ordering_fields = ["created_at", "name", "path"]
ordering = ["-created_at"]
queryset = models.Team.objects.all()
pagination_class = None
def get_queryset(self):
"""Custom queryset to get user related teams."""
teams_queryset = models.Team.objects.filter(
accesses__user=self.request.user,
)
depth_path = teams_queryset.values("depth", "path")
if not depth_path:
return models.Team.objects.none()
user_role_query = models.TeamAccess.objects.filter(
user=self.request.user, team=OuterRef("pk")
).values("role")[:1]
return (
models.Team.objects.prefetch_related("accesses", "service_providers")
.filter(
reduce(
operator.or_,
(
Q(
# The team the user has access to
depth=d["depth"],
path=d["path"],
)
| Q(
# The parent team the user has access to
depth__lt=d["depth"],
path__startswith=d["path"][: models.Team.steplen],
organization_id=self.request.user.organization_id,
)
for d in depth_path
),
),
)
# Abilities are computed based on logged-in user's role for the team
# and if the user does not have access, it's ok to consider them as a member
# because it's a parent team.
.annotate(
user_role=Coalesce(
Subquery(user_role_query), Value(models.RoleChoices.MEMBER.value)
)
)
)
def perform_create(self, serializer):
"""Set the current user as owner of the newly created team."""
team = serializer.save()
models.TeamAccess.objects.create(
team=team,
user=self.request.user,
role=models.RoleChoices.OWNER,
)
class TeamAccessViewSet(
mixins.CreateModelMixin,
mixins.DestroyModelMixin,
mixins.ListModelMixin,
mixins.RetrieveModelMixin,
mixins.UpdateModelMixin,
viewsets.GenericViewSet,
):
"""
API ViewSet for all interactions with team accesses.
GET /api/v1.0/teams/<team_id>/accesses/:<team_access_id>
Return list of all team accesses related to the logged-in user or one
team access if an id is provided.
POST /api/v1.0/teams/<team_id>/accesses/ with expected data:
- user: str
- role: str [owner|admin|member]
Return newly created team access
PUT /api/v1.0/teams/<team_id>/accesses/<team_access_id>/ with expected data:
- role: str [owner|admin|member]
Return updated team access
PATCH /api/v1.0/teams/<team_id>/accesses/<team_access_id>/ with expected data:
- role: str [owner|admin|member]
Return partially updated team access
DELETE /api/v1.0/teams/<team_id>/accesses/<team_access_id>/
Delete targeted team access
"""
lookup_field = "pk"
pagination_class = Pagination
permission_classes = [permissions.AccessPermission]
queryset = (
models.TeamAccess.objects.all().select_related("user").order_by("-created_at")
)
list_serializer_class = serializers.TeamAccessReadOnlySerializer
detail_serializer_class = serializers.TeamAccessSerializer
filter_backends = [filters.OrderingFilter]
ordering = ["role"]
ordering_fields = ["role", "user__email", "user__name"]
def get_permissions(self):
"""User only needs to be authenticated to list team accesses"""
if self.action == "list":
permission_classes = [permissions.IsAuthenticated]
else:
return super().get_permissions()
return [permission() for permission in permission_classes]
def get_serializer_context(self):
"""Extra context provided to the serializer class."""
context = super().get_serializer_context()
context["team_id"] = self.kwargs["team_id"]
return context
def get_serializer_class(self):
"""Chooses list or detail serializer according to the action."""
if self.action in {"list", "retrieve"}:
return self.list_serializer_class
return self.detail_serializer_class
def get_queryset(self):
"""Return the queryset according to the action."""
queryset = super().get_queryset()
queryset = queryset.filter(team=self.kwargs["team_id"])
if self.action in {"list", "retrieve"}:
if query := self.request.GET.get("q", ""):
queryset = queryset.filter(
Q(user__email__unaccent__icontains=query)
| Q(user__name__unaccent__icontains=query)
)
# Determine which role the logged-in user has in the team
user_role_query = models.TeamAccess.objects.filter(
user=self.request.user, team=self.kwargs["team_id"]
).values("role")[:1]
queryset = (
# The logged-in user should be part of a team to see its accesses
queryset.filter(
team__accesses__user=self.request.user,
)
# Abilities are computed based on logged-in user's role and
# the user role on each team access
.annotate(
user_role=Subquery(user_role_query),
)
.select_related("user")
.distinct()
)
return queryset
def destroy(self, request, *args, **kwargs):
"""Forbid deleting the last owner access"""
instance = self.get_object()
team = instance.team
# Check if the access being deleted is the last owner access for the team
if instance.role == "owner" and team.accesses.filter(role="owner").count() == 1:
return response.Response(
{"detail": "Cannot delete the last owner access for the team."},
status=400,
)
return super().destroy(request, *args, **kwargs)
def perform_update(self, serializer):
"""Check that we don't change the role if it leads to losing the last owner."""
instance = serializer.instance
# Check if the role is being updated and the new role is not "owner"
if (
"role" in self.request.data
and self.request.data["role"] != models.RoleChoices.OWNER
):
team = instance.team
# Check if the access being updated is the last owner access for the team
if (
instance.role == models.RoleChoices.OWNER
and team.accesses.filter(role=models.RoleChoices.OWNER).count() == 1
):
message = "Cannot change the role to a non-owner role for the last owner access."
raise exceptions.ValidationError({"role": message})
serializer.save()
class InvitationViewset(
mixins.CreateModelMixin,
mixins.ListModelMixin,
mixins.RetrieveModelMixin,
mixins.DestroyModelMixin,
viewsets.GenericViewSet,
):
"""API ViewSet for user invitations to team.
GET /api/v1.0/teams/<team_id>/invitations/:<invitation_id>/
Return list of invitations related to that team or or one
team access if an id is provided.
POST /api/v1.0/teams/<team_id>/invitations/ with expected data:
- email: str
- role: str [owner|admin|member]
- issuer : User, automatically added from user making query, if allowed
- team : Team, automatically added from requested URI
Return newly created invitation
PUT / PATCH : Not permitted. Instead of updating your invitation,
delete and create a new one.
DELETE /api/v1.0/teams/<team_id>/invitations/<invitation_id>/
Delete targeted invitation
"""
lookup_field = "id"
pagination_class = Pagination
permission_classes = [permissions.AccessPermission]
queryset = (
models.Invitation.objects.all().select_related("team").order_by("-created_at")
)
serializer_class = serializers.InvitationSerializer
def get_permissions(self):
"""User only needs to be authenticated to list invitations"""
if self.action == "list":
permission_classes = [permissions.IsAuthenticated]
else:
return super().get_permissions()
return [permission() for permission in permission_classes]
def get_serializer_context(self):
"""Extra context provided to the serializer class."""
context = super().get_serializer_context()
context["team_id"] = self.kwargs["team_id"]
return context
def get_queryset(self):
"""Return the queryset according to the action."""
queryset = super().get_queryset()
queryset = queryset.filter(team=self.kwargs["team_id"])
if self.action == "list":
# Determine which role the logged-in user has in the team
user_role_query = models.TeamAccess.objects.filter(
user=self.request.user, team=self.kwargs["team_id"]
).values("role")[:1]
queryset = (
# The logged-in user should be part of a team to see its accesses
queryset.filter(
team__accesses__user=self.request.user,
)
# Abilities are computed based on logged-in user's role and
# the user role on each team access
.annotate(user_role=Subquery(user_role_query))
.distinct()
)
return queryset
class ConfigView(views.APIView):
"""API ViewSet for sharing some public settings."""
permission_classes = [AllowAny]
def get(self, request):
"""
GET /api/v1.0/config/
Return a dictionary of public settings.
"""
array_settings = ["LANGUAGES", "FEATURES", "RELEASE", "COMMIT"]
dict_settings = {}
for setting in array_settings:
dict_settings[setting] = getattr(settings, setting)
return response.Response(dict_settings)
class StatView(views.APIView):
"""API ViewSet for sharing some public metrics."""
permission_classes = [AllowAny]
@method_decorator(cache_page(3600))
def get(self, request):
"""
GET /api/v1.0/stats/
Return a dictionary of public metrics.
"""
context = {
"total_users": models.User.objects.count(),
"mau": models.User.objects.filter(
last_login__gte=datetime.datetime.now() - datetime.timedelta(30)
).count(),
"teams": models.Team.objects.count(),
"domains": domains_models.MailDomain.objects.count(),
"mailboxes": domains_models.Mailbox.objects.count(),
}
return response.Response(context)
class ServiceProviderFilter(filters.BaseFilterBackend):
"""
Filter service providers.
"""
def filter_queryset(self, request, queryset, view):
"""
Filter service providers by audience or name.
"""
if name := request.GET.get("name"):
queryset = queryset.filter(name__icontains=name)
if audience_id := request.GET.get("audience_id"):
queryset = queryset.filter(audience_id=audience_id)
return queryset
class ServiceProviderViewSet(
mixins.ListModelMixin,
mixins.RetrieveModelMixin,
viewsets.GenericViewSet,
):
"""
API ViewSet for all interactions with service providers.
GET /api/v1.0/service-providers/
Return a list of service providers.
GET /api/v1.0/service-providers/<service_provider_id>/
Return a service provider.
"""
permission_classes = [permissions.IsAuthenticated]
queryset = models.ServiceProvider.objects.all()
serializer_class = serializers.ServiceProviderSerializer
throttle_classes = [BurstRateThrottle, SustainedRateThrottle]
pagination_class = Pagination
filter_backends = [filters.OrderingFilter, ServiceProviderFilter]
ordering = ["name"]
ordering_fields = ["name", "created_at"]
def get_queryset(self):
"""Filter the queryset to limit results to user's organization."""
queryset = super().get_queryset()
queryset = queryset.filter(organizations__id=self.request.user.organization_id)
return queryset
+2 -18
View File
@@ -1,5 +1,4 @@
"""Permission handlers for the People core app."""
from django.core import exceptions
from rest_framework import permissions
@@ -12,13 +11,13 @@ class IsAuthenticated(permissions.BasePermission):
"""
def has_permission(self, request, view):
"""Check auth token first."""
return bool(request.auth) if request.auth else request.user.is_authenticated
class IsSelf(IsAuthenticated):
"""
Allows access only to user's own data.
Allows access only to authenticated users. Alternative method checking the presence
of the auth token to avoid hitting the database.
"""
def has_object_permission(self, request, view, obj):
@@ -53,18 +52,3 @@ class AccessPermission(IsAuthenticated):
"""Check permission for a given object."""
abilities = obj.get_abilities(request.user)
return abilities.get(request.method.lower(), False)
class TeamPermission(IsAuthenticated):
"""Permission class for team objects viewset."""
def has_permission(self, request, view):
"""Check permission only when the user tries to create a new team."""
if not super().has_permission(request, view):
return False
if request.method != "POST":
return True
abilities = request.user.get_abilities()
return abilities["teams"]["can_create"]
@@ -1 +0,0 @@
"""People core resource server API endpoints"""
@@ -1,52 +0,0 @@
"""Client serializers for the People core app resource server API."""
from rest_framework import serializers
from core import models
class TeamSerializer(serializers.ModelSerializer):
"""Serialize teams."""
class Meta:
model = models.Team
fields = [
"id",
"created_at",
"depth",
"name",
"numchild",
"path",
"updated_at",
]
read_only_fields = [
"id",
"created_at",
"depth",
"numchild",
"path",
"updated_at",
]
def create(self, validated_data):
"""
Create a new team with organization enforcement.
In this context, called as a resource server,
the team service provider is enforced.
When the service provider audience is unknown it is created on the fly.
"""
sp_audience = self.context["from_service_provider_audience"]
service_provider, _created = models.ServiceProvider.objects.get_or_create(
audience_id=sp_audience
)
# Note: this is not the purpose of this API to check the user has an organization
return super().create(
validated_data=validated_data
| {
"organization_id": self.context["request"].user.organization_id,
"service_providers": [service_provider],
},
)
@@ -1,125 +0,0 @@
"""Resource server API endpoints"""
import operator
from functools import reduce
from django.db.models import OuterRef, Prefetch, Q, Subquery, Value
from django.db.models.functions import Coalesce
from rest_framework import (
filters,
mixins,
viewsets,
)
from core import models
from core.api import permissions
from core.api.client.viewsets import Pagination
from core.resource_server.mixins import ResourceServerMixin
from . import serializers
class TeamViewSet( # pylint: disable=too-many-ancestors
ResourceServerMixin,
mixins.CreateModelMixin,
mixins.ListModelMixin,
mixins.RetrieveModelMixin,
mixins.UpdateModelMixin,
viewsets.GenericViewSet,
):
"""
Team ViewSet dedicated to the resource server.
The DELETE method is not allowed for now, because the use case is
not clear yet and it comes with complexity to know if we can delete
a team or not (eg. if a team has other SP, it might not be deleted
but what do we do then, only remove the current SP?).
GET /resource-server/v1.0/teams/
Return list of Teams of the user and available for the audience.
POST /resource-server/v1.0/teams/
Create a new Team for the user for the audience.
GET /resource-server/v1.0/teams/{team_id}/
Return the Team details if available for the audience.
PUT /resource-server/v1.0/teams/{team_id}/
Update the Team details (only name for now).
"""
permission_classes = [permissions.AccessPermission]
serializer_class = serializers.TeamSerializer
filter_backends = [filters.OrderingFilter]
ordering_fields = ["created_at"]
ordering = ["-created_at"]
queryset = models.Team.objects.all()
pagination_class = Pagination
def get_queryset(self):
"""Custom queryset to get user related teams."""
teams_queryset = models.Team.objects.filter(
accesses__user=self.request.user,
)
depth_path = teams_queryset.values("depth", "path")
if not depth_path:
return models.Team.objects.none()
user_role_query = models.TeamAccess.objects.filter(
user=self.request.user, team=OuterRef("pk")
).values("role")[:1]
service_provider_audience = self._get_service_provider_audience()
service_provider_prefetch = Prefetch(
"service_providers",
queryset=models.ServiceProvider.objects.filter(
audience_id=service_provider_audience
),
)
return (
models.Team.objects.prefetch_related(
"accesses",
service_provider_prefetch,
)
.filter(
reduce(
operator.or_,
(
Q(
# The team the user has access to
depth=d["depth"],
path=d["path"],
)
| Q(
# The parent team the user has access to
depth__lt=d["depth"],
path__startswith=d["path"][: models.Team.steplen],
organization_id=self.request.user.organization_id,
)
for d in depth_path
),
),
service_providers__audience_id=service_provider_audience,
)
# Abilities are computed based on logged-in user's role for the team
# and if the user does not have access, it's ok to consider them as a member
# because it's a parent team.
.annotate(
user_role=Coalesce(
Subquery(user_role_query), Value(models.RoleChoices.MEMBER.value)
)
)
)
def perform_create(self, serializer):
"""Set the current user as owner of the newly created team."""
team = serializer.save()
models.TeamAccess.objects.create(
team=team,
user=self.request.user,
role=models.RoleChoices.OWNER,
)
@@ -1,46 +1,30 @@
"""Client serializers for the People core app."""
from rest_framework import exceptions, serializers
from timezone_field.rest_framework import TimeZoneSerializerField
from core import models
from core.models import ServiceProvider
class ContactSerializer(serializers.ModelSerializer):
"""Serialize contacts."""
abilities = serializers.SerializerMethodField()
class Meta:
model = models.Contact
fields = [
"id",
"abilities",
"override",
"base",
"data",
"full_name",
"notes",
"owner",
"short_name",
]
read_only_fields = ["id", "owner", "abilities"]
extra_kwargs = {
"override": {"required": False},
}
read_only_fields = ["id", "owner"]
def update(self, instance, validated_data):
"""Make "override" field readonly but only for update/patch."""
validated_data.pop("override", None)
"""Make "base" field readonly but only for update/patch."""
validated_data.pop("base", None)
return super().update(instance, validated_data)
def get_abilities(self, contact) -> dict:
"""Return abilities of the logged-in user on the instance."""
request = self.context.get("request")
if request:
return contact.get_abilities(request.user)
return {}
class DynamicFieldsModelSerializer(serializers.ModelSerializer):
"""
@@ -49,8 +33,6 @@ class DynamicFieldsModelSerializer(serializers.ModelSerializer):
"""
def __init__(self, *args, **kwargs):
"""Pass arguments to superclass except 'fields', then drop fields not listed therein."""
# Don't pass the 'fields' arg up to the superclass
fields = kwargs.pop("fields", None)
@@ -65,89 +47,45 @@ class DynamicFieldsModelSerializer(serializers.ModelSerializer):
self.fields.pop(field_name)
class OrganizationSerializer(serializers.ModelSerializer):
"""Serialize organizations."""
abilities = serializers.SerializerMethodField()
class Meta:
model = models.Organization
fields = ["id", "name", "registration_id_list", "domain_list", "abilities"]
read_only_fields = ["id", "registration_id_list", "domain_list"]
def get_abilities(self, organization) -> dict:
"""Return abilities of the logged-in user on the instance."""
request = self.context.get("request")
if request:
return organization.get_abilities(request.user)
return {}
class UserOrganizationSerializer(serializers.ModelSerializer):
"""Serialize organizations for users."""
class Meta:
model = models.Organization
fields = ["id", "name", "registration_id_list"]
read_only_fields = ["id", "name", "registration_id_list"]
class UserSerializer(DynamicFieldsModelSerializer):
"""Serialize users."""
timezone = TimeZoneSerializerField(use_pytz=False, required=True)
email = serializers.ReadOnlyField()
name = serializers.ReadOnlyField()
organization = UserOrganizationSerializer(read_only=True)
name = serializers.SerializerMethodField(read_only=True)
email = serializers.SerializerMethodField(read_only=True)
class Meta:
model = models.User
fields = [
"id",
"name",
"email",
"language",
"name",
"organization",
"timezone",
"is_device",
"is_staff",
]
read_only_fields = ["id", "name", "email", "is_device", "is_staff"]
def _get_main_identity_attr(self, obj, attribute_name):
"""Return the specified attribute of the main identity."""
try:
return getattr(obj.main_identity[0], attribute_name)
except TypeError:
return getattr(obj.main_identity, attribute_name)
except IndexError:
main_identity = obj.identities.filter(is_main=True).first()
return getattr(obj.main_identity, attribute_name) if main_identity else None
except AttributeError:
return None
class UserMeSerializer(UserSerializer):
"""
Serialize the current user.
def get_name(self, obj):
"""Return main identity's name."""
return self._get_main_identity_attr(obj, "name")
Same as the `UserSerializer` but with abilities.
"""
abilities = serializers.SerializerMethodField()
organization = UserOrganizationSerializer(read_only=True)
class Meta:
model = models.User
fields = [
"email",
"id",
"is_device",
"is_staff",
"language",
"name",
"organization",
"timezone",
# added fields
"abilities",
]
read_only_fields = ["id", "name", "email", "is_device", "is_staff"]
def get_abilities(self, user: models.User) -> dict:
"""Return abilities of the logged-in user on the instance."""
if user != self.context["request"].user: # Should not happen
raise RuntimeError(
"UserMeSerializer.get_abilities: user is not the same as the request user",
)
return user.get_abilities()
def get_email(self, obj):
"""Return main identity's email."""
return self._get_main_identity_attr(obj, "email")
class TeamAccessSerializer(serializers.ModelSerializer):
@@ -251,50 +189,28 @@ class TeamSerializer(serializers.ModelSerializer):
"""Serialize teams."""
abilities = serializers.SerializerMethodField(read_only=True)
service_providers = serializers.PrimaryKeyRelatedField(
queryset=ServiceProvider.objects.all(), many=True, required=False
)
slug = serializers.SerializerMethodField()
class Meta:
model = models.Team
fields = [
"id",
"abilities",
"accesses",
"created_at",
"depth",
"name",
"numchild",
"path",
"service_providers",
"accesses",
"abilities",
"slug",
"created_at",
"updated_at",
]
read_only_fields = [
"id",
"abilities",
"accesses",
"abilities",
"slug",
"created_at",
"depth",
"numchild",
"path",
"updated_at",
]
def create(self, validated_data):
"""Create a new team with organization enforcement."""
# When called as a resource server, we enforce the team service provider
if sp_audience := self.context.get("from_service_provider_audience", None):
service_provider, _created = models.ServiceProvider.objects.get_or_create(
audience_id=sp_audience
)
validated_data["service_providers"] = [service_provider]
# Note: this is not the purpose of this API to check the user has an organization
return super().create(
validated_data=validated_data
| {"organization_id": self.context["request"].user.organization_id}
)
def get_abilities(self, team) -> dict:
"""Return abilities of the logged-in user on the instance."""
request = self.context.get("request")
@@ -302,46 +218,15 @@ class TeamSerializer(serializers.ModelSerializer):
return team.get_abilities(request.user)
return {}
def get_slug(self, instance):
"""Return slug from the team's name."""
return instance.get_slug()
class InvitationSerializer(serializers.ModelSerializer):
"""Serialize invitations."""
class Meta:
model = models.Invitation
fields = ["id", "created_at", "email", "team", "role", "issuer", "is_expired"]
read_only_fields = ["id", "created_at", "team", "issuer", "is_expired"]
def validate(self, attrs):
"""Validate and restrict invitation to new user based on email."""
request = self.context.get("request")
user = getattr(request, "user", None)
try:
team_id = self.context["team_id"]
except KeyError as exc:
raise exceptions.ValidationError(
"You must set a team ID in kwargs to create a new team invitation."
) from exc
if not models.TeamAccess.objects.filter(
team=team_id,
user=user,
role__in=[models.RoleChoices.OWNER, models.RoleChoices.ADMIN],
).exists():
raise exceptions.PermissionDenied(
"You are not allowed to manage invitation for this team."
)
attrs["team_id"] = team_id
attrs["issuer"] = user
return attrs
class ServiceProviderSerializer(serializers.ModelSerializer):
"""Serialize service providers."""
class Meta:
model = models.ServiceProvider
fields = ["id", "audience_id", "name"]
read_only_fields = ["id", "audience_id"]
fields = ["email", "team", "role", "issuer"]
read_only_fields = ["team", "issuer"]
+401
View File
@@ -0,0 +1,401 @@
"""API endpoints"""
from django.contrib.postgres.search import TrigramSimilarity
from django.db.models import Func, Max, OuterRef, Prefetch, Q, Subquery, Value
from rest_framework import (
decorators,
exceptions,
filters,
mixins,
pagination,
response,
viewsets,
)
from rest_framework.throttling import UserRateThrottle
from core import models
from . import permissions, serializers
EMAIL_SIMILARITY_THRESHOLD = 0.01
# TrigramSimilarity threshold is lower for searching email than for names,
# to improve matching results
class NestedGenericViewSet(viewsets.GenericViewSet):
"""
A generic Viewset aims to be used in a nested route context.
e.g: `/api/v1.0/resource_1/<resource_1_pk>/resource_2/<resource_2_pk>/`
It allows to define all url kwargs and lookup fields to perform the lookup.
"""
lookup_fields: list[str] = ["pk"]
lookup_url_kwargs: list[str] = []
def __getattribute__(self, item):
"""
This method is overridden to allow to get the last lookup field or lookup url kwarg
when accessing the `lookup_field` or `lookup_url_kwarg` attribute. This is useful
to keep compatibility with all methods used by the parent class `GenericViewSet`.
"""
if item in ["lookup_field", "lookup_url_kwarg"]:
return getattr(self, item + "s", [None])[-1]
return super().__getattribute__(item)
def get_queryset(self):
"""
Get the list of items for this view.
`lookup_fields` attribute is enumerated here to perform the nested lookup.
"""
queryset = super().get_queryset()
# The last lookup field is removed to perform the nested lookup as it corresponds
# to the object pk, it is used within get_object method.
lookup_url_kwargs = (
self.lookup_url_kwargs[:-1]
if self.lookup_url_kwargs
else self.lookup_fields[:-1]
)
filter_kwargs = {}
for index, lookup_url_kwarg in enumerate(lookup_url_kwargs):
if lookup_url_kwarg not in self.kwargs:
raise KeyError(
f"Expected view {self.__class__.__name__} to be called with a URL "
f'keyword argument named "{lookup_url_kwarg}". Fix your URL conf, or '
"set the `.lookup_fields` attribute on the view correctly."
)
filter_kwargs.update(
{self.lookup_fields[index]: self.kwargs[lookup_url_kwarg]}
)
return queryset.filter(**filter_kwargs)
class SerializerPerActionMixin:
"""
A mixin to allow to define serializer classes for each action.
This mixin is useful to avoid to define a serializer class for each action in the
`get_serializer_class` method.
"""
serializer_classes: dict[str, type] = {}
default_serializer_class: type = None
def get_serializer_class(self):
"""
Return the serializer class to use depending on the action.
"""
return self.serializer_classes.get(self.action, self.default_serializer_class)
class Pagination(pagination.PageNumberPagination):
"""Pagination to display no more than 100 objects per page sorted by creation date."""
ordering = "-created_on"
max_page_size = 100
page_size_query_param = "page_size"
class BurstRateThrottle(UserRateThrottle):
"""
Throttle rate for minutes. See DRF section in settings for default value.
"""
scope = "burst"
class SustainedRateThrottle(UserRateThrottle):
"""
Throttle rate for hours. See DRF section in settings for default value.
"""
scope = "sustained"
# pylint: disable=too-many-ancestors
class ContactViewSet(
mixins.CreateModelMixin,
mixins.DestroyModelMixin,
mixins.RetrieveModelMixin,
mixins.UpdateModelMixin,
viewsets.GenericViewSet,
):
"""Contact ViewSet"""
permission_classes = [permissions.IsOwnedOrPublic]
queryset = models.Contact.objects.all()
serializer_class = serializers.ContactSerializer
throttle_classes = [BurstRateThrottle, SustainedRateThrottle]
def list(self, request, *args, **kwargs):
"""Limit listed users by a query with throttle protection."""
user = self.request.user
queryset = self.filter_queryset(self.get_queryset())
# Exclude contacts that:
queryset = queryset.filter(
# - belong to another user (keep public and owned contacts)
Q(owner__isnull=True) | Q(owner=user),
# - are profile contacts for a user
user__isnull=True,
# - are overriden base contacts
overriding_contacts__isnull=True,
)
# Search by case-insensitive and accent-insensitive trigram similarity
if query := self.request.GET.get("q", ""):
query = Func(Value(query), function="unaccent")
similarity = TrigramSimilarity(
Func("full_name", function="unaccent"),
query,
) + TrigramSimilarity(Func("short_name", function="unaccent"), query)
queryset = (
queryset.annotate(similarity=similarity)
.filter(
similarity__gte=0.05
) # Value determined by testing (test_api_contacts.py)
.order_by("-similarity")
)
serializer = self.get_serializer(queryset, many=True)
return response.Response(serializer.data)
def perform_create(self, serializer):
"""Set the current user as owner of the newly created contact."""
user = self.request.user
serializer.validated_data["owner"] = user
return super().perform_create(serializer)
class UserViewSet(
mixins.UpdateModelMixin, viewsets.GenericViewSet, mixins.ListModelMixin
):
"""
User viewset for all interactions with user infos and teams.
GET /api/users/&q=query
Return a list of users whose email matches the query. Similarity is
calculated using trigram similarity, allowing for partial,
case-insensitive matches and accented queries.
"""
permission_classes = [permissions.IsSelf]
queryset = models.User.objects.all()
serializer_class = serializers.UserSerializer
throttle_classes = [BurstRateThrottle, SustainedRateThrottle]
pagination_class = Pagination
def get_queryset(self):
"""Limit listed users by a query. Pagination and throttle protection apply."""
queryset = self.queryset
if self.action == "list":
# Exclude inactive contacts
queryset = queryset.filter(
is_active=True,
).prefetch_related(
Prefetch(
"identities",
queryset=models.Identity.objects.filter(is_main=True),
to_attr="main_identity",
)
)
# Search by case-insensitive and accent-insensitive trigram similarity
if query := self.request.GET.get("q", ""):
similarity = Max(
TrigramSimilarity(
Func("identities__email", function="unaccent"),
Func(Value(query), function="unaccent"),
)
)
queryset = (
queryset.annotate(similarity=similarity)
.filter(similarity__gte=EMAIL_SIMILARITY_THRESHOLD)
.order_by("-similarity")
)
return queryset
@decorators.action(
detail=False,
methods=["get"],
url_name="me",
url_path="me",
)
def get_me(self, request):
"""
Return information on currently logged user
"""
user = request.user
user.main_identity = models.Identity.objects.filter(
user=user, is_main=True
).first()
return response.Response(
self.serializer_class(user, context={"request": request}).data
)
class TeamViewSet(
mixins.CreateModelMixin,
mixins.DestroyModelMixin,
mixins.ListModelMixin,
mixins.RetrieveModelMixin,
mixins.UpdateModelMixin,
viewsets.GenericViewSet,
):
"""Team ViewSet"""
permission_classes = [permissions.AccessPermission]
serializer_class = serializers.TeamSerializer
filter_backends = [filters.OrderingFilter]
ordering_fields = ["created_at"]
ordering = ["-created_at"]
queryset = models.Team.objects.all()
def get_queryset(self):
"""Custom queryset to get user related teams."""
user_role_query = models.TeamAccess.objects.filter(
user=self.request.user, team=OuterRef("pk")
).values("role")[:1]
return models.Team.objects.filter(accesses__user=self.request.user).annotate(
user_role=Subquery(user_role_query)
)
def perform_create(self, serializer):
"""Set the current user as owner of the newly created team."""
team = serializer.save()
models.TeamAccess.objects.create(
team=team,
user=self.request.user,
role=models.RoleChoices.OWNER,
)
class TeamAccessViewSet(
mixins.CreateModelMixin,
mixins.DestroyModelMixin,
mixins.ListModelMixin,
mixins.RetrieveModelMixin,
mixins.UpdateModelMixin,
viewsets.GenericViewSet,
):
"""
API ViewSet for all interactions with team accesses.
GET /api/v1.0/teams/<team_id>/accesses/:<team_access_id>
Return list of all team accesses related to the logged-in user or one
team access if an id is provided.
POST /api/v1.0/teams/<team_id>/accesses/ with expected data:
- user: str
- role: str [owner|admin|member]
Return newly created team access
PUT /api/v1.0/teams/<team_id>/accesses/<team_access_id>/ with expected data:
- role: str [owner|admin|member]
Return updated team access
PATCH /api/v1.0/teams/<team_id>/accesses/<team_access_id>/ with expected data:
- role: str [owner|admin|member]
Return partially updated team access
DELETE /api/v1.0/teams/<team_id>/accesses/<team_access_id>/
Delete targeted team access
"""
lookup_field = "pk"
pagination_class = Pagination
permission_classes = [permissions.AccessPermission]
queryset = models.TeamAccess.objects.all().select_related("user")
list_serializer_class = serializers.TeamAccessReadOnlySerializer
detail_serializer_class = serializers.TeamAccessSerializer
def get_permissions(self):
"""User only needs to be authenticated to list team accesses"""
if self.action == "list":
permission_classes = [permissions.IsAuthenticated]
else:
return super().get_permissions()
return [permission() for permission in permission_classes]
def get_serializer_context(self):
"""Extra context provided to the serializer class."""
context = super().get_serializer_context()
context["team_id"] = self.kwargs["team_id"]
return context
def get_serializer_class(self):
if self.action in {"list", "retrieve"}:
return self.list_serializer_class
return self.detail_serializer_class
def get_queryset(self):
"""Return the queryset according to the action."""
queryset = super().get_queryset()
queryset = queryset.filter(team=self.kwargs["team_id"])
if self.action in {"list", "retrieve"}:
# Limit to team access instances related to a team THAT also has a team access
# instance for the logged-in user (we don't want to list only the team access
# instances pointing to the logged-in user)
user_role_query = models.TeamAccess.objects.filter(
team__accesses__user=self.request.user
).values("role")[:1]
queryset = (
queryset.filter(
team__accesses__user=self.request.user,
)
.prefetch_related(
Prefetch(
"user__identities",
queryset=models.Identity.objects.filter(is_main=True),
to_attr="main_identity",
)
)
.annotate(user_role=Subquery(user_role_query))
.distinct()
)
return queryset
def destroy(self, request, *args, **kwargs):
"""Forbid deleting the last owner access"""
instance = self.get_object()
team = instance.team
# Check if the access being deleted is the last owner access for the team
if instance.role == "owner" and team.accesses.filter(role="owner").count() == 1:
return response.Response(
{"detail": "Cannot delete the last owner access for the team."},
status=400,
)
return super().destroy(request, *args, **kwargs)
def perform_update(self, serializer):
"""Check that we don't change the role if it leads to losing the last owner."""
instance = serializer.instance
# Check if the role is being updated and the new role is not "owner"
if (
"role" in self.request.data
and self.request.data["role"] != models.RoleChoices.OWNER
):
team = instance.team
# Check if the access being updated is the last owner access for the team
if (
instance.role == models.RoleChoices.OWNER
and team.accesses.filter(role=models.RoleChoices.OWNER).count() == 1
):
message = "Cannot change the role to a non-owner role for the last owner access."
raise exceptions.ValidationError({"role": message})
serializer.save()
+121
View File
@@ -0,0 +1,121 @@
"""Authentication for the People core app."""
from django.conf import settings
from django.core.exceptions import SuspiciousOperation
from django.db import models
from django.utils.translation import gettext_lazy as _
import requests
from mozilla_django_oidc.auth import (
OIDCAuthenticationBackend as MozillaOIDCAuthenticationBackend,
)
from .models import Identity
class OIDCAuthenticationBackend(MozillaOIDCAuthenticationBackend):
"""Custom OpenID Connect (OIDC) Authentication Backend.
This class overrides the default OIDC Authentication Backend to accommodate differences
in the User and Identity models, and handles signed and/or encrypted UserInfo response.
"""
def get_userinfo(self, access_token, id_token, payload):
"""Return user details dictionary.
Parameters:
- access_token (str): The access token.
- id_token (str): The id token (unused).
- payload (dict): The token payload (unused).
Note: The id_token and payload parameters are unused in this implementation,
but were kept to preserve base method signature.
Note: It handles signed and/or encrypted UserInfo Response. It is required by
Agent Connect, which follows the OIDC standard. It forces us to override the
base method, which deal with 'application/json' response.
Returns:
- dict: User details dictionary obtained from the OpenID Connect user endpoint.
"""
user_response = requests.get(
self.OIDC_OP_USER_ENDPOINT,
headers={"Authorization": f"Bearer {access_token}"},
verify=self.get_settings("OIDC_VERIFY_SSL", True),
timeout=self.get_settings("OIDC_TIMEOUT", None),
proxies=self.get_settings("OIDC_PROXY", None),
)
user_response.raise_for_status()
userinfo = self.verify_token(user_response.text)
return userinfo
def get_or_create_user(self, access_token, id_token, payload):
"""Return a User based on userinfo. Get or create a new user if no user matches the Sub.
Parameters:
- access_token (str): The access token.
- id_token (str): The ID token.
- payload (dict): The user payload.
Returns:
- User: An existing or newly created User instance.
Raises:
- Exception: Raised when user creation is not allowed and no existing user is found.
"""
user_info = self.get_userinfo(access_token, id_token, payload)
# Compute user name from OIDC name fields as defined in settings
names_list = [
user_info[field]
for field in settings.USER_OIDC_FIELDS_TO_NAME
if user_info.get(field)
]
user_info["name"] = " ".join(names_list) or None
sub = user_info.get("sub")
if sub is None:
raise SuspiciousOperation(
_("User info contained no recognizable user identification")
)
user = (
self.UserModel.objects.filter(identities__sub=sub)
.annotate(
identity_email=models.F("identities__email"),
identity_name=models.F("identities__name"),
)
.distinct()
.first()
)
if user:
email = user_info.get("email")
name = user_info.get("name")
if (
email
and email != user.identity_email
or name
and name != user.identity_name
):
Identity.objects.filter(sub=sub).update(email=email, name=name)
elif self.get_settings("OIDC_CREATE_USER", True):
user = self.create_user(user_info)
return user
def create_user(self, claims):
"""Return a newly created User instance."""
sub = claims.get("sub")
if sub is None:
raise SuspiciousOperation(
_("Claims contained no recognizable user identification")
)
user = self.UserModel.objects.create(password="!") # noqa: S106
Identity.objects.create(
user=user, sub=sub, email=claims.get("email"), name=claims.get("name")
)
return user
@@ -1 +0,0 @@
"""Authentication module."""
-248
View File
@@ -1,248 +0,0 @@
"""Authentication Backends for the People core app."""
import logging
from email.headerregistry import Address
from typing import Optional
from django.conf import settings
from django.contrib.auth import get_user_model
from django.core.exceptions import SuspiciousOperation
from django.utils.translation import gettext_lazy as _
import requests
from mozilla_django_oidc.auth import (
OIDCAuthenticationBackend as MozillaOIDCAuthenticationBackend,
)
from core.models import (
Contact,
Organization,
OrganizationAccess,
OrganizationRoleChoices,
)
logger = logging.getLogger(__name__)
User = get_user_model()
def get_domain_from_email(email: Optional[str]) -> Optional[str]:
"""Extract domain from email."""
try:
return Address(addr_spec=email).domain
except (ValueError, AttributeError):
return None
class OIDCAuthenticationBackend(MozillaOIDCAuthenticationBackend):
"""Custom OpenID Connect (OIDC) Authentication Backend.
This class overrides the default OIDC Authentication Backend to accommodate differences
in the User model, and handles signed and/or encrypted UserInfo response.
"""
def get_userinfo(self, access_token, id_token, payload):
"""Return user details dictionary.
Parameters:
- access_token (str): The access token.
- id_token (str): The id token (unused).
- payload (dict): The token payload (unused).
Note: The id_token and payload parameters are unused in this implementation,
but were kept to preserve base method signature.
Note: It handles signed and/or encrypted UserInfo Response. It is required by
Agent Connect, which follows the OIDC standard. It forces us to override the
base method, which deal with 'application/json' response.
Returns:
- dict: User details dictionary obtained from the OpenID Connect user endpoint.
"""
user_response = requests.get(
self.OIDC_OP_USER_ENDPOINT,
headers={"Authorization": f"Bearer {access_token}"},
verify=self.get_settings("OIDC_VERIFY_SSL", True),
timeout=self.get_settings("OIDC_TIMEOUT", None),
proxies=self.get_settings("OIDC_PROXY", None),
)
user_response.raise_for_status()
userinfo = self.verify_token(user_response.text)
return userinfo
def get_or_create_user(self, access_token, id_token, payload):
"""Return a User based on userinfo. Create a new user if no match is found.
Parameters:
- access_token (str): The access token.
- id_token (str): The ID token.
- payload (dict): The user payload.
Returns:
- User: An existing or newly created User instance.
Raises:
- Exception: Raised when user creation is not allowed and no existing user is found.
"""
user_info = self.get_userinfo(access_token, id_token, payload)
sub = user_info.get("sub")
if not sub:
raise SuspiciousOperation(
_("User info contained no recognizable user identification")
)
# Get user's full name from OIDC fields defined in settings
full_name = self.compute_full_name(user_info)
email = user_info.get("email")
claims = {
"sub": sub,
"email": email,
"name": full_name,
}
if settings.OIDC_ORGANIZATION_REGISTRATION_ID_FIELD:
claims[settings.OIDC_ORGANIZATION_REGISTRATION_ID_FIELD] = user_info.get(
settings.OIDC_ORGANIZATION_REGISTRATION_ID_FIELD
)
# if sub is absent, try matching on email
user = self.get_existing_user(sub, email)
if user:
if not user.is_active:
raise SuspiciousOperation(_("User account is disabled"))
self.update_user_if_needed(user, claims)
elif self.get_settings("OIDC_CREATE_USER", True):
user = self.create_user(claims)
# Data cleaning, to be removed when user organization is null=False
# or all users have an organization.
# See https://github.com/numerique-gouv/people/issues/504
if not user.organization_id:
organization_registration_id = claims.get(
settings.OIDC_ORGANIZATION_REGISTRATION_ID_FIELD
)
domain = get_domain_from_email(email)
try:
organization, organization_created = (
Organization.objects.get_or_create_from_user_claims(
registration_id=organization_registration_id,
domain=domain,
)
)
if organization_created:
logger.info("Organization %s created", organization)
# For this case, we don't create an OrganizationAccess we will
# manage this manually later, because we don't want the first
# user who log in after the release to be the admin of their
# organization. We will keep organization without admin, and
# we will have to manually clean things up (while there is
# not that much organization in the database).
except ValueError as exc:
# Raised when there is no recognizable organization
# identifier (domain or registration_id)
logger.warning("Unable to update user organization: %s", exc)
else:
user.organization = organization
user.save()
logger.info(
"User %s updated with organization %s", user.pk, organization
)
return user
def create_user(self, claims):
"""Return a newly created User instance."""
sub = claims.get("sub")
if sub is None:
raise SuspiciousOperation(
_("Claims contained no recognizable user identification")
)
email = claims.get("email")
name = claims.get("name")
# Extract or create the organization from the data
organization_registration_id = claims.get(
settings.OIDC_ORGANIZATION_REGISTRATION_ID_FIELD
)
domain = get_domain_from_email(email)
try:
organization, organization_created = (
Organization.objects.get_or_create_from_user_claims(
registration_id=organization_registration_id,
domain=domain,
)
)
except ValueError as exc:
raise SuspiciousOperation(
_("Claims contained no recognizable organization identification")
) from exc
if organization_created:
logger.info("Organization %s created", organization)
logger.info("Creating user %s / %s", sub, email)
user = self.UserModel.objects.create(
organization=organization,
password="!", # noqa: S106
sub=sub,
email=email,
name=name,
)
if organization_created:
# Warning: we may remove this behavior in the near future when we
# add a feature to claim the organization ownership.
OrganizationAccess.objects.create(
organization=organization,
user=user,
role=OrganizationRoleChoices.ADMIN,
)
# Initiate the user's profile
Contact.objects.create(
owner=user,
user=user,
full_name=name or email,
data={
"emails": [
{"type": "Work", "value": email},
],
},
)
return user
def compute_full_name(self, user_info):
"""Compute user's full name based on OIDC fields in settings."""
name_fields = settings.USER_OIDC_FIELDS_TO_NAME
full_name = " ".join(
user_info[field] for field in name_fields if user_info.get(field)
)
return full_name or None
def get_existing_user(self, sub, email):
"""Fetch existing user by sub or email."""
try:
return User.objects.get(sub=sub)
except User.DoesNotExist:
if email and settings.OIDC_FALLBACK_TO_EMAIL_FOR_IDENTIFICATION:
try:
return User.objects.get(email=email)
except User.DoesNotExist:
pass
return None
def update_user_if_needed(self, user, claims):
"""Update user claims if they have changed."""
updated_claims = {}
for key in ["email", "name"]:
claim_value = claims.get(key)
if claim_value and claim_value != getattr(user, key):
updated_claims[key] = claim_value
if updated_claims:
self.UserModel.objects.filter(sub=user.sub).update(**updated_claims)
-18
View File
@@ -1,18 +0,0 @@
"""Authentication URLs for the People core app."""
from django.urls import path
from mozilla_django_oidc.urls import urlpatterns as mozzila_oidc_urls
from .views import OIDCLogoutCallbackView, OIDCLogoutView
urlpatterns = [
# Override the default 'logout/' path from Mozilla Django OIDC with our custom view.
path("logout/", OIDCLogoutView.as_view(), name="oidc_logout_custom"),
path(
"logout-callback/",
OIDCLogoutCallbackView.as_view(),
name="oidc_logout_callback",
),
*mozzila_oidc_urls,
]
-137
View File
@@ -1,137 +0,0 @@
"""Authentication Views for the People core app."""
from urllib.parse import urlencode
from django.contrib import auth
from django.core.exceptions import SuspiciousOperation
from django.http import HttpResponseRedirect
from django.urls import reverse
from django.utils import crypto
from mozilla_django_oidc.utils import (
absolutify,
)
from mozilla_django_oidc.views import (
OIDCLogoutView as MozillaOIDCOIDCLogoutView,
)
class OIDCLogoutView(MozillaOIDCOIDCLogoutView):
"""Custom logout view for handling OpenID Connect (OIDC) logout flow.
Adds support for handling logout callbacks from the identity provider (OP)
by initiating the logout flow if the user has an active session.
The Django session is retained during the logout process to persist the 'state' OIDC parameter.
This parameter is crucial for maintaining the integrity of the logout flow between this call
and the subsequent callback.
"""
@staticmethod
def persist_state(request, state):
"""Persist the given 'state' parameter in the session's 'oidc_states' dictionary
This method is used to store the OIDC state parameter in the session, according to the
structure expected by Mozilla Django OIDC's 'add_state_and_verifier_and_nonce_to_session'
utility function.
"""
if "oidc_states" not in request.session or not isinstance(
request.session["oidc_states"], dict
):
request.session["oidc_states"] = {}
request.session["oidc_states"][state] = {}
request.session.save()
def construct_oidc_logout_url(self, request):
"""Create the redirect URL for interfacing with the OIDC provider.
Retrieves the necessary parameters from the session and constructs the URL
required to initiate logout with the OpenID Connect provider.
If no ID token is found in the session, the logout flow will not be initiated,
and the method will return the default redirect URL.
The 'state' parameter is generated randomly and persisted in the session to ensure
its integrity during the subsequent callback.
"""
oidc_logout_endpoint = self.get_settings("OIDC_OP_LOGOUT_ENDPOINT")
if not oidc_logout_endpoint:
return self.redirect_url
reverse_url = reverse("oidc_logout_callback")
id_token = request.session.get("oidc_id_token", None)
if not id_token:
return self.redirect_url
query = {
"id_token_hint": id_token,
"state": crypto.get_random_string(self.get_settings("OIDC_STATE_SIZE", 32)),
"post_logout_redirect_uri": absolutify(request, reverse_url),
}
self.persist_state(request, query["state"])
return f"{oidc_logout_endpoint}?{urlencode(query)}"
def post(self, request):
"""Handle user logout.
If the user is not authenticated, redirects to the default logout URL.
Otherwise, constructs the OIDC logout URL and redirects the user to start
the logout process.
If the user is redirected to the default logout URL, ensure her Django session
is terminated.
"""
logout_url = self.redirect_url
if request.user.is_authenticated:
logout_url = self.construct_oidc_logout_url(request)
# If the user is not redirected to the OIDC provider, ensure logout
if logout_url == self.redirect_url:
auth.logout(request)
return HttpResponseRedirect(logout_url)
class OIDCLogoutCallbackView(MozillaOIDCOIDCLogoutView):
"""Custom view for handling the logout callback from the OpenID Connect (OIDC) provider.
Handles the callback after logout from the identity provider (OP).
Verifies the state parameter and performs necessary logout actions.
The Django session is maintained during the logout process to ensure the integrity
of the logout flow initiated in the previous step.
"""
http_method_names = ["get"]
def get(self, request):
"""Handle the logout callback.
If the user is not authenticated, redirects to the default logout URL.
Otherwise, verifies the state parameter and performs necessary logout actions.
"""
if not request.user.is_authenticated:
return HttpResponseRedirect(self.redirect_url)
state = request.GET.get("state")
if state not in request.session.get("oidc_states", {}):
msg = "OIDC callback state not found in session `oidc_states`!"
raise SuspiciousOperation(msg)
del request.session["oidc_states"][state]
request.session.save()
auth.logout(request)
return HttpResponseRedirect(self.redirect_url)
-10
View File
@@ -1,9 +1,7 @@
"""
Core application enums declaration
"""
from django.conf import global_settings, settings
from django.db import models
from django.utils.translation import gettext_lazy as _
# Django sets `LANGUAGES` by default with all supported languages. We can use it for
@@ -15,11 +13,3 @@ ALL_LANGUAGES = getattr(
"ALL_LANGUAGES",
[(language, _(name)) for language, name in global_settings.LANGUAGES],
)
class WebhookStatusChoices(models.TextChoices):
"""Defines the possible statuses in which a webhook can be."""
FAILURE = "failure", _("Failure")
PENDING = "pending", _("Pending")
SUCCESS = "success", _("Success")
+23 -117
View File
@@ -2,7 +2,6 @@
"""
Core application factories
"""
from django.conf import settings
from django.contrib.auth.hashers import make_password
@@ -115,83 +114,32 @@ class ContactFactory(BaseContactFactory):
class Meta:
model = models.Contact
owner = factory.SubFactory("core.factories.UserFactory")
class ProfileContactFactory(BaseContactFactory):
"""A factory to create profile contacts for a user"""
class Meta:
model = models.Contact
owner = factory.SelfAttribute(".user")
user = factory.SubFactory("core.factories.UserFactory")
class OverrideContactFactory(BaseContactFactory):
"""A factory to create contacts for a user"""
class Meta:
model = models.Contact
override = factory.SubFactory("core.factories.ContactFactory", owner=None)
owner = factory.SubFactory("core.factories.UserFactory")
class OrganizationFactory(factory.django.DjangoModelFactory):
"""Factory to create organizations for testing purposes."""
name = factory.Faker("company")
class Meta:
model = models.Organization
class Params: # pylint: disable=missing-class-docstring
with_registration_id = factory.Trait(
registration_id_list=factory.List(
[factory.Sequence(lambda n: f"{n:014d}")]
),
)
with_domain = factory.Trait(
domain_list=factory.List([factory.Faker("domain_name")]),
)
class OrganizationAccessFactory(factory.django.DjangoModelFactory):
"""Factory to create organization accesses for testing purposes."""
class Meta:
model = models.OrganizationAccess
user = factory.SubFactory(
"core.factories.UserFactory",
organization=factory.SelfAttribute("..organization"),
)
organization = factory.SubFactory(
"core.factories.OrganizationFactory", with_registration_id=True
)
role = factory.fuzzy.FuzzyChoice(models.OrganizationRoleChoices.values)
base = factory.SubFactory("core.factories.ContactFactory", base=None, owner=None)
owner = factory.SubFactory("core.factories.UserFactory", profile_contact=None)
class UserFactory(factory.django.DjangoModelFactory):
"""A factory to create random users for testing purposes."""
"""A factory to random users for testing purposes."""
class Meta:
model = models.User
email = factory.Faker("email")
language = factory.fuzzy.FuzzyChoice([lang[0] for lang in settings.LANGUAGES])
password = make_password("password")
class IdentityFactory(factory.django.DjangoModelFactory):
"""A factory to create identities for a user"""
class Meta:
model = models.Identity
django_get_or_create = ("sub",)
class Params:
with_organization = factory.Trait(
organization=factory.SubFactory(
OrganizationFactory, with_registration_id=True
),
)
user = factory.SubFactory(UserFactory)
sub = factory.Sequence(lambda n: f"user{n!s}")
email = factory.Faker("email")
name = factory.Faker("name")
language = factory.fuzzy.FuzzyChoice([lang[0] for lang in settings.LANGUAGES])
password = make_password("password")
class TeamFactory(factory.django.DjangoModelFactory):
@@ -199,27 +147,19 @@ class TeamFactory(factory.django.DjangoModelFactory):
class Meta:
model = models.Team
skip_postgeneration_save = True
django_get_or_create = ("name",)
name = factory.Sequence(lambda n: f"team{n}")
@factory.post_generation
def users(self, create, extracted, **kwargs):
"""Add users to team from a given list of users with or without roles."""
if not create or not extracted:
return
for user_entry in extracted:
if isinstance(user_entry, models.User):
TeamAccessFactory(team=self, user=user_entry)
else:
TeamAccessFactory(team=self, user=user_entry[0], role=user_entry[1])
@factory.post_generation
def service_providers(self, create, extracted, **kwargs):
"""Add service providers to team from a given list of service providers."""
if not create or not extracted:
return
self.service_providers.set(extracted)
if create and extracted:
for item in extracted:
if isinstance(item, models.User):
TeamAccessFactory(team=self, user=item)
else:
TeamAccessFactory(team=self, user=item[0], role=item[1])
class TeamAccessFactory(factory.django.DjangoModelFactory):
@@ -233,47 +173,13 @@ class TeamAccessFactory(factory.django.DjangoModelFactory):
role = factory.fuzzy.FuzzyChoice([r[0] for r in models.RoleChoices.choices])
class TeamWebhookFactory(factory.django.DjangoModelFactory):
"""Create fake team webhooks for testing."""
class Meta:
model = models.TeamWebhook
team = factory.SubFactory(TeamFactory)
url = factory.Sequence(lambda n: f"https://example.com/Groups/{n!s}")
class InvitationFactory(factory.django.DjangoModelFactory):
"""A factory to create invitations for a user"""
class Meta:
model = models.Invitation
team = factory.SubFactory(TeamFactory)
email = factory.Faker("email")
team = factory.SubFactory(TeamFactory)
role = factory.fuzzy.FuzzyChoice([role[0] for role in models.RoleChoices.choices])
issuer = factory.SubFactory(UserFactory)
class ServiceProviderFactory(factory.django.DjangoModelFactory):
"""A factory to create service providers for testing purposes."""
class Meta:
model = models.ServiceProvider
skip_postgeneration_save = True
audience_id = factory.Faker("uuid4")
@factory.post_generation
def teams(self, create, extracted, **kwargs):
"""Add teams to service provider from a given list."""
if not create or not extracted:
return
self.teams.set(extracted)
@factory.post_generation
def organizations(self, create, extracted, **kwargs):
"""Add organization to service provider from a given list."""
if not create or not extracted:
return
self.organizations.set(extracted)
+29 -31
View File
@@ -1,4 +1,4 @@
# Generated by Django 5.0.3 on 2024-06-08 09:25
# Generated by Django 5.0.2 on 2024-02-23 06:46
import django.contrib.auth.models
import django.core.validators
@@ -45,10 +45,8 @@ class Migration(migrations.Migration):
('id', models.UUIDField(default=uuid.uuid4, editable=False, help_text='primary key for the record as UUID', primary_key=True, serialize=False, verbose_name='id')),
('created_at', models.DateTimeField(auto_now_add=True, help_text='date and time at which a record was created', verbose_name='created at')),
('updated_at', models.DateTimeField(auto_now=True, help_text='date and time at which a record was last updated', verbose_name='updated at')),
('sub', models.CharField(help_text='Required. 255 characters or fewer. Letters, numbers, and @/./+/-/_ characters only.', max_length=255, unique=True, validators=[django.core.validators.RegexValidator(message='Enter a valid sub. This value may contain only letters, numbers, and @/./+/-/_ characters.', regex='^[\\w.@+-]+\\Z')], verbose_name='sub')),
('email', models.EmailField(blank=True, max_length=254, null=True, verbose_name='email address')),
('name', models.CharField(blank=True, max_length=100, null=True, verbose_name='name')),
('language', models.CharField(choices=settings.LANGUAGES, default='en-us', help_text='The language in which the user wants to see the interface.', max_length=10, verbose_name='language')),
('email', models.EmailField(blank=True, max_length=254, null=True, unique=True, verbose_name='email address')),
('language', models.CharField(choices="(('en-us', 'English'), ('fr-fr', 'French'))", default='en-us', help_text='The language in which the user wants to see the interface.', max_length=10, verbose_name='language')),
('timezone', timezone_field.fields.TimeZoneField(choices_display='WITH_GMT_OFFSET', default='UTC', help_text='The timezone in which the user wants to see times.', use_pytz=False)),
('is_device', models.BooleanField(default=False, help_text='Whether the user is a device or a real user.', verbose_name='device')),
('is_staff', models.BooleanField(default=False, help_text='Whether the user can log into this admin site.', verbose_name='staff status')),
@@ -89,6 +87,25 @@ class Migration(migrations.Migration):
name='profile_contact',
field=models.OneToOneField(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='user', to='core.contact'),
),
migrations.CreateModel(
name='Identity',
fields=[
('id', models.UUIDField(default=uuid.uuid4, editable=False, help_text='primary key for the record as UUID', primary_key=True, serialize=False, verbose_name='id')),
('created_at', models.DateTimeField(auto_now_add=True, help_text='date and time at which a record was created', verbose_name='created at')),
('updated_at', models.DateTimeField(auto_now=True, help_text='date and time at which a record was last updated', verbose_name='updated at')),
('sub', models.CharField(help_text='Required. 255 characters or fewer. Letters, numbers, and @/./+/-/_ characters only.', max_length=255, unique=True, validators=[django.core.validators.RegexValidator(message='Enter a valid sub. This value may contain only letters, numbers, and @/./+/-/_ characters.', regex='^[\\w.@+-]+\\Z')], verbose_name='sub')),
('email', models.EmailField(blank=True, max_length=254, null=True, verbose_name='email address')),
('name', models.CharField(blank=True, max_length=100, null=True, verbose_name='name')),
('is_main', models.BooleanField(default=False, help_text='Designates whether the email is the main one.', verbose_name='main')),
('user', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='identities', to=settings.AUTH_USER_MODEL)),
],
options={
'verbose_name': 'identity',
'verbose_name_plural': 'identities',
'db_table': 'people_identity',
'ordering': ('-is_main', 'email'),
},
),
migrations.CreateModel(
name='Invitation',
fields=[
@@ -101,9 +118,7 @@ class Migration(migrations.Migration):
('team', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='invitations', to='core.team')),
],
options={
'verbose_name': 'Team invitation',
'verbose_name_plural': 'Team invitations',
'db_table': 'people_invitation',
'abstract': False,
},
),
migrations.CreateModel(
@@ -127,38 +142,21 @@ class Migration(migrations.Migration):
name='users',
field=models.ManyToManyField(related_name='teams', through='core.TeamAccess', to=settings.AUTH_USER_MODEL),
),
migrations.CreateModel(
name='TeamWebhook',
fields=[
('id', models.UUIDField(default=uuid.uuid4, editable=False, help_text='primary key for the record as UUID', primary_key=True, serialize=False, verbose_name='id')),
('created_at', models.DateTimeField(auto_now_add=True, help_text='date and time at which a record was created', verbose_name='created at')),
('updated_at', models.DateTimeField(auto_now=True, help_text='date and time at which a record was last updated', verbose_name='updated at')),
('url', models.URLField(verbose_name='url')),
('secret', models.CharField(blank=True, max_length=255, null=True, verbose_name='secret')),
('status', models.CharField(choices=[('failure', 'Failure'), ('pending', 'Pending'), ('success', 'Success')], default='pending', max_length=10)),
('team', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='webhooks', to='core.team')),
],
options={
'verbose_name': 'Team webhook',
'verbose_name_plural': 'Team webhooks',
'db_table': 'people_team_webhook',
},
migrations.AddConstraint(
model_name='contact',
constraint=models.CheckConstraint(check=models.Q(('base__isnull', False), ('owner__isnull', True), _negated=True), name='base_owner_constraint', violation_error_message='A contact overriding a base contact must be owned.'),
),
migrations.AddConstraint(
model_name='contact',
constraint=models.CheckConstraint(condition=models.Q(('base__isnull', False), ('owner__isnull', True), _negated=True), name='base_owner_constraint', violation_error_message='A contact overriding a base contact must be owned.'),
),
migrations.AddConstraint(
model_name='contact',
constraint=models.CheckConstraint(condition=models.Q(('base', models.F('id')), _negated=True), name='base_not_self', violation_error_message='A contact cannot be based on itself.'),
constraint=models.CheckConstraint(check=models.Q(('base', models.F('id')), _negated=True), name='base_not_self', violation_error_message='A contact cannot be based on itself.'),
),
migrations.AlterUniqueTogether(
name='contact',
unique_together={('owner', 'base')},
),
migrations.AddConstraint(
model_name='invitation',
constraint=models.UniqueConstraint(fields=('email', 'team'), name='email_and_team_unique_together'),
model_name='identity',
constraint=models.UniqueConstraint(fields=('user', 'email'), name='unique_user_email', violation_error_message='This email address is already declared for this user.'),
),
migrations.AddConstraint(
model_name='teamaccess',
@@ -1,62 +0,0 @@
# Generated by Django 5.1.1 on 2024-10-22 10:07
import core.models
import django.contrib.postgres.fields
import django.db.models.deletion
import uuid
from django.conf import settings
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('core', '0001_initial'),
]
operations = [
migrations.CreateModel(
name='Organization',
fields=[
('id', models.UUIDField(default=uuid.uuid4, editable=False, help_text='primary key for the record as UUID', primary_key=True, serialize=False, verbose_name='id')),
('created_at', models.DateTimeField(auto_now_add=True, help_text='date and time at which a record was created', verbose_name='created at')),
('updated_at', models.DateTimeField(auto_now=True, help_text='date and time at which a record was last updated', verbose_name='updated at')),
('name', models.CharField(max_length=100, verbose_name='name')),
('registration_id_list', django.contrib.postgres.fields.ArrayField(base_field=models.CharField(max_length=128), blank=True, default=list, size=None, verbose_name='registration ID list')),
('domain_list', django.contrib.postgres.fields.ArrayField(base_field=models.CharField(max_length=256), blank=True, default=list, size=None, verbose_name='domain list')),
],
options={
'verbose_name': 'organization',
'verbose_name_plural': 'organizations',
'db_table': 'people_organization',
'constraints': [models.CheckConstraint(condition=models.Q(('registration_id_list__len__gt', 0), ('domain_list__len__gt', 0), _connector='OR'), name='registration_id_or_domain', violation_error_message='An organization must have at least a registration ID or a domain.')],
},
),
migrations.AddField(
model_name='team',
name='organization',
field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.PROTECT, related_name='teams', to='core.organization'),
),
migrations.AddField(
model_name='user',
name='organization',
field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.PROTECT, related_name='users', to='core.organization'),
),
migrations.CreateModel(
name='OrganizationAccess',
fields=[
('id', models.UUIDField(default=uuid.uuid4, editable=False, help_text='primary key for the record as UUID', primary_key=True, serialize=False, verbose_name='id')),
('created_at', models.DateTimeField(auto_now_add=True, help_text='date and time at which a record was created', verbose_name='created at')),
('updated_at', models.DateTimeField(auto_now=True, help_text='date and time at which a record was last updated', verbose_name='updated at')),
('role', models.CharField(choices=[('administrator', 'Administrator')], default='administrator', max_length=20)),
('organization', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='organization_accesses', to='core.organization')),
('user', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='organization_accesses', to=settings.AUTH_USER_MODEL)),
],
options={
'verbose_name': 'Organization/user relation',
'verbose_name_plural': 'Organization/user relations',
'db_table': 'people_organization_access',
'constraints': [models.UniqueConstraint(fields=('user', 'organization'), name='unique_organization_user', violation_error_message='This user is already in this organization.')],
},
),
]
@@ -1,18 +0,0 @@
# Generated by Django 5.1.2 on 2024-11-04 14:25
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('core', '0002_add_organization_and_more'),
]
operations = [
migrations.AlterField(
model_name='team',
name='slug',
field=models.SlugField(max_length=100, null=True),
),
]
@@ -1,17 +0,0 @@
# Generated by Django 5.1.3 on 2024-11-20 10:08
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('core', '0003_team_slug_nullable'),
]
operations = [
migrations.RemoveField(
model_name='team',
name='slug',
),
]
@@ -1,39 +0,0 @@
# Generated by Django 5.1.2 on 2024-11-07 16:24
import uuid
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('core', '0004_remove_team_slug'),
]
operations = [
migrations.CreateModel(
name='ServiceProvider',
fields=[
('id', models.UUIDField(default=uuid.uuid4, editable=False, help_text='primary key for the record as UUID', primary_key=True, serialize=False, verbose_name='id')),
('created_at', models.DateTimeField(auto_now_add=True, help_text='date and time at which a record was created', verbose_name='created at')),
('updated_at', models.DateTimeField(auto_now=True, help_text='date and time at which a record was last updated', verbose_name='updated at')),
('name', models.CharField(max_length=256, unique=True, verbose_name='name')),
('audience_id', models.CharField(db_index=True, max_length=256, unique=True, verbose_name='audience id')),
],
options={
'verbose_name': 'service provider',
'verbose_name_plural': 'service providers',
'db_table': 'people_service_provider',
},
),
migrations.AddField(
model_name='organization',
name='service_providers',
field=models.ManyToManyField(blank=True, related_name='organizations', to='core.serviceprovider'),
),
migrations.AddField(
model_name='team',
name='service_providers',
field=models.ManyToManyField(blank=True, related_name='teams', to='core.serviceprovider'),
),
]
@@ -1,24 +0,0 @@
# Generated by Django 5.1.3 on 2024-11-28 09:37
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('core', '0005_add_serviceprovider'),
]
operations = [
migrations.AddField(
model_name='contact',
name='notes',
field=models.TextField(blank=True, default='', verbose_name='notes'),
),
migrations.AlterField(
model_name='contact',
name='full_name',
field=models.CharField(default='-', max_length=150, verbose_name='full name'),
preserve_default=False,
),
]
@@ -1,47 +0,0 @@
# Generated by Django 5.1.3 on 2024-12-02 10:04
import django.db.models.deletion
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('core', '0006_contact_notes_alter_contact_full_name'),
]
operations = [
migrations.RemoveConstraint(
model_name='contact',
name='base_owner_constraint',
),
migrations.RemoveConstraint(
model_name='contact',
name='base_not_self',
),
migrations.AlterUniqueTogether(
name='contact',
unique_together=set(),
),
migrations.AddField(
model_name='contact',
name='override',
field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='overridden_by', to='core.contact'),
),
migrations.AddConstraint(
model_name='contact',
constraint=models.CheckConstraint(condition=models.Q(('override__isnull', False), ('owner__isnull', True), _negated=True), name='override_owner_constraint', violation_error_message='A contact overriding a base contact must be owned.'),
),
migrations.AddConstraint(
model_name='contact',
constraint=models.CheckConstraint(condition=models.Q(('override', models.F('id')), _negated=True), name='override_not_self', violation_error_message='A contact cannot override itself.'),
),
migrations.RemoveField(
model_name='contact',
name='base',
),
migrations.AlterUniqueTogether(
name='contact',
unique_together={('owner', 'override')},
),
]
@@ -1,47 +0,0 @@
# Generated by Django 5.1.3 on 2024-12-02 10:58
import django.db.models.deletion
from django.conf import settings
from django.db import migrations, models
def init_profile_contact(apps, schema_editor):
"""Create a Contact for each User to be used as their profile."""
User = apps.get_model("core", "User")
Contact = apps.get_model("core", "Contact")
for user in User.objects.all(): # This is a small table for now
Contact.objects.create(
owner=user,
user=user,
full_name=user.name or user.email,
data={
"emails": [
{"type": "Work", "value": user.email},
],
},
)
class Migration(migrations.Migration):
dependencies = [
('core', '0007_rename_contact_base_to_override'),
]
operations = [
migrations.RemoveField(
model_name='user',
name='profile_contact',
),
migrations.AddField(
model_name='contact',
name='user',
field=models.OneToOneField(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='profile_contact', to=settings.AUTH_USER_MODEL),
),
migrations.AddConstraint(
model_name='contact',
constraint=models.CheckConstraint(condition=models.Q(('user__isnull', True), models.Q(('owner', models.F('user')), ('owner__isnull', False)), _connector='OR'), name='profile_contact_owner_constraint', violation_error_message='Users can only declare as profile a contact they own.'),
),
migrations.RunPython(init_profile_contact, reverse_code=migrations.RunPython.noop),
]
@@ -1,18 +0,0 @@
# Generated by Django 5.1.3 on 2024-12-05 13:19
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('core', '0008_change_user_profile_to_contact'),
]
operations = [
# From 100ms to 78ms for 5 000 contacts on my computer
# This is not a big improvement and may need further investigation
migrations.RunSQL(
"CREATE INDEX json_field_index_emails_value ON people_contact USING GIN ((data->'emails'->'value') jsonb_path_ops);"
)
]
@@ -1,58 +0,0 @@
# Generated by Django 5.1.3 on 2024-11-26 13:55
from django.db import migrations, models
from treebeard.numconv import NumConv
def update_team_paths(apps, schema_editor):
Team = apps.get_model("core", "Team")
alphabet = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz"
steplen = 5
# Initialize NumConv with the specified custom alphabet
converter = NumConv(len(alphabet), alphabet)
nodes = Team.objects.all().order_by("created_at")
for i, node in enumerate(nodes, 1):
node.depth = 1 # root nodes are at depth 1
# Use NumConv to encode the index `i` to a base representation and
# pad it to the specified step length using the custom alphabet
node.path = converter.int2str(i).rjust(steplen, alphabet[0])
if nodes:
Team.objects.bulk_update(nodes, ["depth", "path"])
class Migration(migrations.Migration):
dependencies = [
('core', '0009_contacts_add_index_on_data_emails'),
]
operations = [
migrations.AddField(
model_name='team',
name='depth',
field=models.PositiveIntegerField(default=0),
preserve_default=False,
),
migrations.AddField(
model_name='team',
name='numchild',
field=models.PositiveIntegerField(default=0),
),
migrations.AddField(
model_name='team',
name='path',
field=models.CharField(default='', max_length=400, db_collation="C"),
preserve_default=False,
),
migrations.RunPython(update_team_paths, migrations.RunPython.noop),
migrations.AlterField(
model_name="team",
name="path",
field=models.CharField(unique=True, max_length=400, db_collation="C"),
),
]
+113 -622
View File
@@ -1,39 +1,23 @@
"""
Declare and configure the models for the People core application
"""
import json
import os
import smtplib
import uuid
from contextlib import suppress
from datetime import timedelta
from logging import getLogger
from typing import Tuple
from django.conf import settings
from django.contrib.auth import models as auth_models
from django.contrib.auth.base_user import AbstractBaseUser
from django.contrib.postgres.fields import ArrayField
from django.contrib.sites.models import Site
from django.core import exceptions, mail, validators
from django.core.exceptions import ValidationError
from django.db import models, transaction
from django.template.loader import render_to_string
from django.db import models
from django.utils import timezone
from django.utils.translation import gettext, override
from django.utils.functional import lazy
from django.utils.text import slugify
from django.utils.translation import gettext_lazy as _
import jsonschema
from timezone_field import TimeZoneField
from treebeard.mp_tree import MP_Node, MP_NodeManager
from core.enums import WebhookStatusChoices
from core.plugins.loader import organization_plugins_run_after_create
from core.utils.webhooks import scim_synchronizer
from core.validators import get_field_validators_from_setting
logger = getLogger(__name__)
current_dir = os.path.dirname(os.path.abspath(__file__))
contact_schema_path = os.path.join(current_dir, "jsonschema", "contact_data.json")
@@ -41,7 +25,7 @@ with open(contact_schema_path, "r", encoding="utf-8") as contact_schema_file:
contact_schema = json.load(contact_schema_file)
class RoleChoices(models.TextChoices): # pylint: disable=too-many-ancestors
class RoleChoices(models.TextChoices):
"""Defines the possible roles a user can have in a team."""
MEMBER = "member", _("Member")
@@ -49,17 +33,6 @@ class RoleChoices(models.TextChoices): # pylint: disable=too-many-ancestors
OWNER = "owner", _("Owner")
class OrganizationRoleChoices(models.TextChoices): # pylint: disable=too-many-ancestors
"""
Defines the possible roles a user can have in an organization.
For now, we only have one role, but we might add more in the future.
administrator: The user can manage the organization: change name, add/remove users.
"""
ADMIN = "administrator", _("Administrator")
class BaseModel(models.Model):
"""
Serves as an abstract base model for other models, ensuring that records are validated
@@ -94,16 +67,16 @@ class BaseModel(models.Model):
def save(self, *args, **kwargs):
"""Call `full_clean` before saving."""
self.full_clean()
return super().save(*args, **kwargs)
super().save(*args, **kwargs)
class Contact(BaseModel):
"""User contacts"""
override = models.ForeignKey(
base = models.ForeignKey(
"self",
on_delete=models.SET_NULL,
related_name="overridden_by",
related_name="overriding_contacts",
null=True,
blank=True,
)
@@ -114,19 +87,11 @@ class Contact(BaseModel):
null=True,
blank=True,
)
user = models.OneToOneField(
settings.AUTH_USER_MODEL,
on_delete=models.SET_NULL,
related_name="profile_contact",
blank=True,
null=True,
)
full_name = models.CharField(_("full name"), max_length=150)
full_name = models.CharField(_("full name"), max_length=150, null=True, blank=True)
short_name = models.CharField(_("short name"), max_length=30, null=True, blank=True)
# avatar =
notes = models.TextField(_("notes"), blank=True, default="")
# notes =
data = models.JSONField(
_("contact information"),
help_text=_("A JSON object containing the contact information"),
@@ -145,38 +110,31 @@ class Contact(BaseModel):
ordering = ("full_name", "short_name")
verbose_name = _("contact")
verbose_name_plural = _("contacts")
unique_together = ("owner", "override")
unique_together = ("owner", "base")
constraints = [
models.CheckConstraint(
condition=~models.Q(override__isnull=False, owner__isnull=True),
name="override_owner_constraint",
check=~models.Q(base__isnull=False, owner__isnull=True),
name="base_owner_constraint",
violation_error_message="A contact overriding a base contact must be owned.",
),
models.CheckConstraint(
condition=~models.Q(override=models.F("id")),
name="override_not_self",
violation_error_message="A contact cannot override itself.",
),
# When a user is set, the owner must be the user
models.CheckConstraint(
condition=models.Q(user__isnull=True)
| models.Q(owner=models.F("user"), owner__isnull=False),
name="profile_contact_owner_constraint",
violation_error_message="Users can only declare as profile a contact they own.",
check=~models.Q(base=models.F("id")),
name="base_not_self",
violation_error_message="A contact cannot be based on itself.",
),
]
def __str__(self):
return self.full_name
return self.full_name or self.short_name
def clean(self):
"""Validate fields."""
super().clean()
# Check if the contact overrides a contact that itself overrides another base contact
if self.override_id and self.override.override_id:
# Check if the contact points to a base contact that itself points to another base contact
if self.base_id and self.base.base_id:
raise exceptions.ValidationError(
"A contact cannot override a contact that itself overrides a contact."
"A contact cannot point to a base contact that itself points to a base contact."
)
# Validate the content of the "data" field against our jsonschema definition
@@ -188,260 +146,21 @@ class Contact(BaseModel):
error_message = f"Validation error in '{field_path:s}': {e.message}"
raise exceptions.ValidationError({"data": [error_message]}) from e
def get_abilities(self, user):
"""
Compute and return abilities for a given user on the contact.
Beware that the model allows owner to be None, we are still not
sure about the use case for this and the API does not allow this.
For now, we still consider here, a contact without owner is "public"
so we allow access to it.
"""
is_owner = user.pk == self.owner_id
is_profile_member_or_same_organization = bool(self.user_id) and (
self.user.organization_id == user.organization_id
)
return {
"get": is_owner or is_profile_member_or_same_organization or not self.owner,
"patch": is_owner,
"put": is_owner,
"delete": is_owner and not self.user, # Can't delete a profile contact
}
class ServiceProvider(BaseModel):
"""
Represents a service provider that will consume our information.
Organization uses this model to define the list of SP available to their users.
Team uses this model to define their visibility to the various SP.
"""
name = models.CharField(_("name"), max_length=256, unique=True)
audience_id = models.CharField(
_("audience id"), max_length=256, unique=True, db_index=True
)
class Meta:
db_table = "people_service_provider"
verbose_name = _("service provider")
verbose_name_plural = _("service providers")
def __str__(self):
return self.name
def save(self, *args, **kwargs):
"""Enforce name (even if ugly) from the `audience_id` field."""
if not self.name:
self.name = self.audience_id # ok, same length
return super().save(*args, **kwargs)
class OrganizationManager(models.Manager):
"""
Custom manager for the Organization model, to manage complexity/automation.
"""
def get_or_create_from_user_claims(
self, registration_id: str = None, domain: str = None, **kwargs
) -> Tuple["Organization", bool]:
"""
Get or create an organization using the most fitting information from the user's claims.
We expect to have only one organization per registration_id, but
the registration_id might not be provided.
When the registration_id is not provided, we use the domain to identify the organization.
If both are provided, we use the registration_id first to create missing organization.
Dev note: When a registration_id is provided by the Identity Provider, we don't want
to use the domain to create the organization, because it is less reliable: for example,
a professional user, may have a personal email address, and the domain would be gmail.com
which is not a good identifier for an organization. The domain email is just a fallback
when the registration_id is not provided by the Identity Provider. We can use the domain
to create the organization manually when we are sure about the "safety" of it.
"""
if not any([registration_id, domain]):
raise ValueError("You must provide either a registration_id or a domain.")
filters = models.Q()
if registration_id:
filters |= models.Q(registration_id_list__icontains=registration_id)
if domain:
filters |= models.Q(domain_list__icontains=domain)
with suppress(self.model.DoesNotExist):
# If there are several organizations, we must raise an error and fix the data
# If there is an organization, we return it
return self.get(filters, **kwargs), False
# Manage the case where the organization does not exist: we create one
if registration_id:
return self.create(
name=registration_id, registration_id_list=[registration_id], **kwargs
), True
if domain:
return self.create(name=domain, domain_list=[domain], **kwargs), True
raise ValueError("Should never reach this point.")
def create(self, **kwargs):
"""
Create an organization with the given kwargs.
This method is overridden to call the Organization plugins.
"""
instance = super().create(**kwargs)
organization_plugins_run_after_create(instance)
return instance
class Organization(BaseModel):
"""
Organization model used to regroup Teams.
Each User have an Organization, which corresponds actually to a default organization
because a user can belong to a Team from another organization.
Each Team have an Organization, which is the Organization from the User who created
the Team.
Organization is managed automatically, the User should never choose their Organization.
When creating a User, you must use the `get_or_create` method from the
OrganizationManager to find the proper Organization.
An Organization can have several registration IDs and domains but during automatic
creation process, only one will be used. We may want to allow (manual) organization merge
later, to regroup several registration IDs or domain in the same Organization.
"""
name = models.CharField(_("name"), max_length=100)
registration_id_list = ArrayField(
models.CharField(
max_length=128,
validators=get_field_validators_from_setting(
"ORGANIZATION_REGISTRATION_ID_VALIDATORS"
),
),
verbose_name=_("registration ID list"),
default=list,
blank=True,
# list overlap validation is done in the validate_unique method
)
domain_list = ArrayField(
models.CharField(max_length=256),
verbose_name=_("domain list"),
default=list,
blank=True,
# list overlap validation is done in the validate_unique method
)
service_providers = models.ManyToManyField(
ServiceProvider,
related_name="organizations",
blank=True,
)
objects = OrganizationManager()
class Meta:
db_table = "people_organization"
verbose_name = _("organization")
verbose_name_plural = _("organizations")
constraints = [
models.CheckConstraint(
name="registration_id_or_domain",
condition=models.Q(registration_id_list__len__gt=0)
| models.Q(domain_list__len__gt=0),
violation_error_message=_(
"An organization must have at least a registration ID or a domain."
),
),
# Check a registration ID str can only be present in one
# organization registration ID list
# Check a domain str can only be present in one organization domain list
# Those checks cannot be done with Django constraints
]
def __str__(self):
return f"{self.name} (# {self.pk})"
def validate_unique(self, exclude=None):
"""
Validate Registration/Domain values in an array field are unique
across all instances.
This can't be done in the field validator because we need to
exclude the current object if already in database.
"""
super().validate_unique(exclude=exclude)
if self.pk:
organization_qs = Organization.objects.exclude(pk=self.pk)
else:
organization_qs = Organization.objects.all()
# Check a registration ID can only be present in one organization
if (
self.registration_id_list
and organization_qs.filter(
registration_id_list__overlap=self.registration_id_list
).exists()
):
raise ValidationError(
"registration_id_list value must be unique across all instances."
)
# Check a domain can only be present in one organization
if (
self.domain_list
and organization_qs.filter(domain_list__overlap=self.domain_list).exists()
):
raise ValidationError(
"domain_list value must be unique across all instances."
)
def get_abilities(self, user):
"""
Compute and return abilities for a given user on the organization.
"""
# Use the role from queryset annotation will raise on purpose if not used properly
is_admin = self.user_role == OrganizationRoleChoices.ADMIN # pylint: disable=no-member
return {
"get": user.organization_id == self.pk,
"patch": is_admin,
"put": is_admin,
"delete": False,
}
class User(AbstractBaseUser, BaseModel, auth_models.PermissionsMixin):
"""User model to work with OIDC only authentication."""
sub_validator = validators.RegexValidator(
regex=r"^[\w.@+-]+\Z",
message=_(
"Enter a valid sub. This value may contain only letters, "
"numbers, and @/./+/-/_ characters."
),
email = models.EmailField(_("email address"), unique=True, null=True, blank=True)
profile_contact = models.OneToOneField(
Contact,
on_delete=models.SET_NULL,
related_name="user",
blank=True,
null=True,
)
sub = models.CharField(
_("sub"),
help_text=_(
"Required. 255 characters or fewer. Letters, numbers, and @/./+/-/_ characters only."
),
max_length=255,
unique=True,
validators=[sub_validator],
)
email = models.EmailField(_("email address"), null=True, blank=True)
name = models.CharField(_("name"), max_length=100, null=True, blank=True)
language = models.CharField(
max_length=10,
choices=settings.LANGUAGES,
choices=lazy(lambda: settings.LANGUAGES, tuple)(),
default=settings.LANGUAGE_CODE,
verbose_name=_("language"),
help_text=_("The language in which the user wants to see the interface."),
@@ -470,17 +189,10 @@ class User(AbstractBaseUser, BaseModel, auth_models.PermissionsMixin):
"Unselect this instead of deleting accounts."
),
)
organization = models.ForeignKey(
Organization,
on_delete=models.PROTECT,
related_name="users",
null=True, # Need to be set to False when everything is migrated
blank=True, # Need to be set to False when everything is migrated
)
objects = auth_models.UserManager()
USERNAME_FIELD = "sub"
USERNAME_FIELD = "email"
REQUIRED_FIELDS = []
class Meta:
@@ -489,195 +201,114 @@ class User(AbstractBaseUser, BaseModel, auth_models.PermissionsMixin):
verbose_name_plural = _("users")
def __str__(self):
return self.name if self.name else self.email or f"User {self.sub}"
def save(self, *args, **kwargs):
"""
If it's a new user, give her access to the relevant teams.
"""
if self._state.adding:
self._convert_valid_invitations()
super().save(*args, **kwargs)
return (
str(self.profile_contact)
if self.profile_contact
else self.email or str(self.id)
)
def clean(self):
"""Validate fields."""
super().clean()
if self.email:
self.email = User.objects.normalize_email(self.email)
def _convert_valid_invitations(self):
"""
Convert valid invitations to team accesses.
Expired invitations are ignored.
"""
valid_invitations = Invitation.objects.filter(
email=self.email,
created_at__gte=(
timezone.now()
- timedelta(seconds=settings.INVITATION_VALIDITY_DURATION)
),
).select_related("team")
if not valid_invitations.exists():
return
TeamAccess.objects.bulk_create(
[
TeamAccess(user=self, team=invitation.team, role=invitation.role)
for invitation in valid_invitations
]
)
valid_invitations.delete()
if self.profile_contact_id and not self.profile_contact.owner == self:
raise exceptions.ValidationError(
"Users can only declare as profile a contact they own."
)
def email_user(self, subject, message, from_email=None, **kwargs):
"""Email this user."""
if not self.email:
raise ValueError("You must first set an email for the user.")
mail.send_mail(subject, message, from_email, [self.email], **kwargs)
main_identity = self.identities.get(is_main=True)
mail.send_mail(subject, message, from_email, [main_identity.email], **kwargs)
def get_abilities(self):
@classmethod
def get_email_field_name(cls):
"""
Return the user permissions at the application level (ie feature availability).
Note: this is for display purposes only for now.
- Contacts view and creation is globally available (or not) for now.
- Teams view is available if the user is owner or admin of at least
one Team for now. It allows to restrict users knowing the existence of
this feature.
- Teams creation is globally available (or not) for now.
- Mailboxes view is available if the user has access to at least one
MailDomain for now. It allows to restrict users knowing the existence of
this feature.
- Mailboxes creation is globally available (or not) for now.
Raise error when trying to get email field name from the user as we are using
a separate Email model to allow several emails per user.
"""
user_info = (
self._meta.model.objects.filter(pk=self.pk)
.annotate(
teams_can_view=models.Exists(
self.accesses.model.objects.filter( # pylint: disable=no-member
user=models.OuterRef("pk"),
role__in=[RoleChoices.OWNER, RoleChoices.ADMIN],
)
),
mailboxes_can_view=models.Exists(
self.mail_domain_accesses.model.objects.filter( # pylint: disable=no-member
user=models.OuterRef("pk")
)
),
)
.only("id")
.get()
raise NotImplementedError(
"This feature is deactivated to allow several emails per user."
)
teams_can_view = user_info.teams_can_view or settings.FEATURES["TEAMS_DISPLAY"]
mailboxes_can_view = user_info.mailboxes_can_view
return {
"contacts": {
"can_view": settings.FEATURES["CONTACTS_DISPLAY"],
"can_create": (
settings.FEATURES["CONTACTS_DISPLAY"]
and settings.FEATURES["CONTACTS_CREATE"]
),
},
"teams": {
"can_view": teams_can_view,
"can_create": teams_can_view and settings.FEATURES["TEAMS_CREATE"],
},
"mailboxes": {
"can_view": mailboxes_can_view,
"can_create": mailboxes_can_view
and settings.FEATURES["MAILBOXES_CREATE"],
},
}
class Identity(BaseModel):
"""User identity"""
class OrganizationAccess(BaseModel):
"""
Link table between organization and users,
only for user with specific rights on Organization.
"""
organization = models.ForeignKey(
Organization,
on_delete=models.CASCADE,
related_name="organization_accesses",
sub_validator = validators.RegexValidator(
regex=r"^[\w.@+-]+\Z",
message=_(
"Enter a valid sub. This value may contain only letters, "
"numbers, and @/./+/-/_ characters."
),
)
user = models.ForeignKey(
User,
on_delete=models.CASCADE,
related_name="organization_accesses",
user = models.ForeignKey(User, related_name="identities", on_delete=models.CASCADE)
sub = models.CharField(
_("sub"),
help_text=_(
"Required. 255 characters or fewer. Letters, numbers, and @/./+/-/_ characters only."
),
max_length=255,
unique=True,
validators=[sub_validator],
)
role = models.CharField(
max_length=20,
choices=OrganizationRoleChoices.choices,
default=OrganizationRoleChoices.ADMIN,
email = models.EmailField(_("email address"), null=True, blank=True)
name = models.CharField(_("name"), max_length=100, null=True, blank=True)
is_main = models.BooleanField(
_("main"),
default=False,
help_text=_("Designates whether the email is the main one."),
)
class Meta:
db_table = "people_organization_access"
verbose_name = _("Organization/user relation")
verbose_name_plural = _("Organization/user relations")
db_table = "people_identity"
ordering = ("-is_main", "email")
verbose_name = _("identity")
verbose_name_plural = _("identities")
constraints = [
# Uniqueness
models.UniqueConstraint(
fields=["user", "organization"],
name="unique_organization_user",
violation_error_message=_("This user is already in this organization."),
fields=["user", "email"],
name="unique_user_email",
violation_error_message=_(
"This email address is already declared for this user."
),
),
]
def __str__(self):
return f"{self.user!s} is {self.role:s} in organization {self.organization!s}"
main_str = "[main]" if self.is_main else ""
id_str = self.email or self.sub
return f"{id_str:s}{main_str:s}"
def save(self, *args, **kwargs):
"""Ensure users always have one and only one main identity."""
super().save(*args, **kwargs)
if self.is_main is True:
self.user.identities.exclude(id=self.id).update(is_main=False)
def clean(self):
"""Normalize the email field and clean the 'is_main' field."""
if self.email:
self.email = User.objects.normalize_email(self.email)
if not self.user.identities.exclude(pk=self.pk).filter(is_main=True).exists():
if not self.created_at:
self.is_main = True
elif not self.is_main:
raise exceptions.ValidationError(
{"is_main": "A user should have one and only one main identity."}
)
super().clean()
class TeamManager(MP_NodeManager):
"""
Custom manager for the Team model, to manage complexity/automation.
"""
def create(self, parent_id=None, **kwargs):
"""
Replace the default create method to ease the Team creation process.
Notes:
- the `add_*` methods from django-treebeard does not support the "using db".
Which means it will always use the default db.
- the `add_*` methods from django-treebeard does not support the "force_insert".
"""
if parent_id is None:
return self.model.add_root(**kwargs)
# Retrieve parent object, because django-treebeard uses raw queries for most
# write operations, and raw queries dont update the django objects of the db
# entries they modify. See caveats in the django-treebeard documentation.
# This might be changed later if we never do any operation on the parent object
# before creating the child.
# Beware the N+1 here.
return self.get(pk=parent_id).add_child(**kwargs)
class Team(MP_Node, BaseModel):
class Team(BaseModel):
"""
Represents the link between teams and users, specifying the role a user has in a team.
When a team is created from here, the user have to choose which Service Providers
can see it.
When a team is created from a Service Provider this one is automatically set in the
Team `service_providers`.
"""
# Allow up to 80 nested teams with 62^5 (916_132_832) root nodes
alphabet = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz"
# Django treebeard does not allow max_length = None...
steplen = 5
path = models.CharField(max_length=5 * 80, unique=True, db_collation="C")
name = models.CharField(max_length=100)
slug = models.SlugField(max_length=100, unique=True, null=False, editable=False)
users = models.ManyToManyField(
User,
@@ -685,20 +316,6 @@ class Team(MP_Node, BaseModel):
through_fields=("team", "user"),
related_name="teams",
)
organization = models.ForeignKey(
Organization,
on_delete=models.PROTECT,
related_name="teams",
null=True, # Need to be set to False when everything is migrated
blank=True, # Need to be set to False when everything is migrated
)
service_providers = models.ManyToManyField(
ServiceProvider,
related_name="teams",
blank=True,
)
objects = TeamManager()
class Meta:
db_table = "people_team"
@@ -709,6 +326,15 @@ class Team(MP_Node, BaseModel):
def __str__(self):
return self.name
def save(self, *args, **kwargs):
"""Overriding save function to compute the slug."""
self.slug = self.get_slug()
return super().save(*args, **kwargs)
def get_slug(self):
"""Compute slug value from name."""
return slugify(self.name)
def get_abilities(self, user):
"""
Compute and return abilities for a given user on the team.
@@ -728,7 +354,7 @@ class Team(MP_Node, BaseModel):
is_owner_or_admin = role in [RoleChoices.OWNER, RoleChoices.ADMIN]
return {
"get": bool(role),
"get": True,
"patch": is_owner_or_admin,
"put": is_owner_or_admin,
"delete": role == RoleChoices.OWNER,
@@ -768,34 +394,6 @@ class TeamAccess(BaseModel):
def __str__(self):
return f"{self.user!s} is {self.role:s} in team {self.team!s}"
def save(self, *args, **kwargs):
"""
Override save function to fire webhooks on any addition or update
to a team access.
"""
if self._state.adding:
self.team.webhooks.update(status=WebhookStatusChoices.PENDING)
with transaction.atomic():
instance = super().save(*args, **kwargs)
scim_synchronizer.add_user_to_group(self.team, self.user)
else:
instance = super().save(*args, **kwargs)
return instance
def delete(self, *args, **kwargs):
"""
Override delete method to fire webhooks on to team accesses.
Don't allow deleting a team access until it is successfully synchronized with all
its webhooks.
"""
self.team.webhooks.update(status=WebhookStatusChoices.PENDING)
with transaction.atomic():
arguments = self.team, self.user
super().delete(*args, **kwargs)
scim_synchronizer.remove_user_from_group(*arguments)
def get_abilities(self, user):
"""
Compute and return abilities for a given user taking into account
@@ -846,34 +444,6 @@ class TeamAccess(BaseModel):
}
class TeamWebhook(BaseModel):
"""Webhooks fired on changes in teams."""
team = models.ForeignKey(Team, related_name="webhooks", on_delete=models.CASCADE)
url = models.URLField(_("url"))
secret = models.CharField(_("secret"), max_length=255, null=True, blank=True)
status = models.CharField(
max_length=10,
default=WebhookStatusChoices.PENDING,
choices=WebhookStatusChoices.choices,
)
class Meta:
db_table = "people_team_webhook"
verbose_name = _("Team webhook")
verbose_name_plural = _("Team webhooks")
def __str__(self):
return f"Webhook to {self.url} for {self.team}"
def get_headers(self):
"""Build header dict from webhook object."""
headers = {"Content-Type": "application/json"}
if self.secret:
headers["Authorization"] = f"Bearer {self.secret:s}"
return headers
class Invitation(BaseModel):
"""User invitation to teams."""
@@ -892,90 +462,11 @@ class Invitation(BaseModel):
related_name="invitations",
)
class Meta:
db_table = "people_invitation"
verbose_name = _("Team invitation")
verbose_name_plural = _("Team invitations")
constraints = [
models.UniqueConstraint(
fields=["email", "team"], name="email_and_team_unique_together"
)
]
def __str__(self):
return f"{self.email} invited to {self.team}"
def save(self, *args, **kwargs):
"""Make invitations read-only."""
if self.created_at:
raise exceptions.PermissionDenied()
super().save(*args, **kwargs)
self.email_invitation()
def clean(self):
"""Validate fields."""
super().clean()
# Check if a user already exists for the provided email
if User.objects.filter(email=self.email).exists():
raise exceptions.ValidationError(
{"email": _("This email is already associated to a registered user.")}
)
@property
def is_expired(self):
"""Calculate if invitation is still valid or has expired."""
if not self.created_at:
return None
validity_duration = timedelta(seconds=settings.INVITATION_VALIDITY_DURATION)
return timezone.now() > (self.created_at + validity_duration)
def get_abilities(self, user):
"""Compute and return abilities for a given user."""
can_delete = False
role = None
if user.is_authenticated:
try:
role = self.user_role
except AttributeError:
try:
role = self.team.accesses.filter(user=user).values("role")[0][
"role"
]
except (self._meta.model.DoesNotExist, IndexError):
role = None
can_delete = role in [RoleChoices.OWNER, RoleChoices.ADMIN]
return {
"delete": can_delete,
"get": bool(role),
"patch": False,
"put": False,
}
def email_invitation(self):
"""Email invitation to the user."""
try:
with override(self.issuer.language):
subject = gettext("Invitation to join La Régie!")
template_vars = {
"title": subject,
"site": Site.objects.get_current(),
}
msg_html = render_to_string("mail/html/invitation.html", template_vars)
msg_plain = render_to_string("mail/text/invitation.txt", template_vars)
mail.send_mail(
subject,
msg_plain,
settings.EMAIL_FROM,
[self.email],
html_message=msg_html,
fail_silently=False,
)
except smtplib.SMTPException as exception:
logger.error("invitation to %s was not sent: %s", self.email, exception)
-1
View File
@@ -1 +0,0 @@
"""Core plugins package."""
-13
View File
@@ -1,13 +0,0 @@
"""Base plugin class for organization plugins."""
class BaseOrganizationPlugin:
"""
Base class for organization plugins.
Plugins must implement all methods of this class even if it is only to "pass".
"""
def run_after_create(self, organization) -> None:
"""Method called after creating an organization."""
raise NotImplementedError("Plugins must implement the run_after_create method")
-32
View File
@@ -1,32 +0,0 @@
"""Helper functions to load and run organization plugins."""
from functools import lru_cache
from typing import List
from django.conf import settings
from django.utils.module_loading import import_string
from core.plugins.base import BaseOrganizationPlugin
@lru_cache(maxsize=None)
def get_organization_plugins() -> List[BaseOrganizationPlugin]:
"""
Return a list of all organization plugins.
While the plugins initialization does not depend on the request, we can cache the result.
"""
return [
import_string(plugin_path)() for plugin_path in settings.ORGANIZATION_PLUGINS
]
def organization_plugins_run_after_create(organization):
"""
Run the after create method for all organization plugins.
Each plugin will be called in the order they are listed in the settings.
Each plugin is responsible to save changes if needed, this is not optimized
but this could be easily improved later if needed.
"""
for plugin_instance in get_organization_plugins():
plugin_instance.run_after_create(organization)
@@ -1 +0,0 @@
"""Backend resource server module."""
@@ -1,72 +0,0 @@
"""Resource Server Authentication"""
import base64
import binascii
import logging
from django.conf import settings
from django.core.exceptions import ImproperlyConfigured
from mozilla_django_oidc.contrib.drf import OIDCAuthentication
from .backend import ResourceServerBackend, ResourceServerImproperlyConfiguredBackend
from .clients import AuthorizationServerClient
logger = logging.getLogger(__name__)
class ResourceServerAuthentication(OIDCAuthentication):
"""Authenticate clients using the token received from the authorization server."""
def __init__(self):
"""Require authentication to be configured in order to instantiate."""
super().__init__()
try:
authorization_server_client = AuthorizationServerClient(
url=settings.OIDC_OP_URL,
verify_ssl=settings.OIDC_VERIFY_SSL,
timeout=settings.OIDC_TIMEOUT,
proxy=settings.OIDC_PROXY,
url_jwks=settings.OIDC_OP_JWKS_ENDPOINT,
url_introspection=settings.OIDC_OP_INTROSPECTION_ENDPOINT,
)
self.backend = ResourceServerBackend(authorization_server_client)
except ImproperlyConfigured as err:
message = "Resource Server authentication is disabled"
logger.debug("%s. Exception: %s", message, err)
self.backend = ResourceServerImproperlyConfiguredBackend()
def get_access_token(self, request):
"""Retrieve and decode the access token from the request.
This method overrides the 'get_access_token' method from the parent class,
to support service providers that would base64 encode the bearer token.
"""
access_token = super().get_access_token(request)
try:
access_token = base64.b64decode(access_token).decode("utf-8")
except (binascii.Error, TypeError):
pass
return access_token
def authenticate(self, request):
"""
Authenticate the request and return a tuple of (user, token) or None.
We override the 'authenticate' method from the parent class to store
the introspected token audience inside the request.
"""
result = super().authenticate(request) # Might raise AuthenticationFailed
if result is None: # Case when there is no access token
return None
# Note: at this stage, the request is a "drf_request" object
request.resource_server_token_audience = self.backend.token_origin_audience
return result
-240
View File
@@ -1,240 +0,0 @@
"""Resource Server Backend"""
import logging
from django.conf import settings
from django.contrib import auth
from django.core.exceptions import ImproperlyConfigured, SuspiciousOperation
from joserfc import jwe as jose_jwe
from joserfc import jwt as jose_jwt
from joserfc.errors import InvalidClaimError, InvalidTokenError
from requests.exceptions import HTTPError
from rest_framework.exceptions import AuthenticationFailed
from . import utils
logger = logging.getLogger(__name__)
class ResourceServerBackend:
"""Backend of an OAuth 2.0 resource server.
This backend is designed to authenticate resource owners to our API using the access token
they received from the authorization server.
In the context of OAuth 2.0, a resource server is a server that hosts protected resources and
is capable of accepting and responding to protected resource requests using access tokens.
The resource server verifies the validity of the access tokens issued by the authorization
server to ensure secure access to the resources.
For more information, visit: https://www.oauth.com/oauth2-servers/the-resource-server/
"""
# pylint: disable=too-many-instance-attributes
def __init__(self, authorization_server_client):
"""Require client_id, client_secret set and authorization_server_client provided."""
# pylint: disable=invalid-name
self.UserModel = auth.get_user_model()
self._client_id = settings.OIDC_RS_CLIENT_ID
self._client_secret = settings.OIDC_RS_CLIENT_SECRET
self._encryption_encoding = settings.OIDC_RS_ENCRYPTION_ENCODING
self._encryption_algorithm = settings.OIDC_RS_ENCRYPTION_ALGO
self._signing_algorithm = settings.OIDC_RS_SIGNING_ALGO
self._scopes = settings.OIDC_RS_SCOPES
if (
not self._client_id
or not self._client_secret
or not authorization_server_client
):
raise ImproperlyConfigured(
"Could not instantiate ResourceServerBackend, some parameters are missing."
)
self._authorization_server_client = authorization_server_client
self._claims_registry = jose_jwt.JWTClaimsRegistry(
iss={"essential": True, "value": self._authorization_server_client.url},
aud={"essential": True, "value": self._client_id},
token_introspection={"essential": True},
)
# Declare the token origin audience: to know where the token comes from
# and store it for further use in the application
self.token_origin_audience = None
# pylint: disable=unused-argument
def get_or_create_user(self, access_token, id_token, payload):
"""Maintain API compatibility with OIDCAuthentication class from mozilla-django-oidc
Params 'id_token', 'payload' won't be used, and our implementation will only
support 'get_user', not 'get_or_create_user'.
"""
return self.get_user(access_token)
def get_user(self, access_token):
"""Get user from an access token emitted by the authorization server.
This method will submit the access token to the authorization server for
introspection, to ensure its validity and obtain the associated metadata.
It follows the specifications outlined in RFC7662 https://www.rfc-editor.org/info/rfc7662,
https://datatracker.ietf.org/doc/html/draft-ietf-oauth-jwt-introspection-response-12.
In our eGovernment applications, the standard RFC 7662 doesn't provide sufficient security.
Its introspection response is a plain JSON object. Therefore, we use the draft RFC
that extends RFC 7662 by returning a signed and encrypted JWT for stronger assurance that
the authorization server issued the token introspection response.
"""
self.token_origin_audience = None # Reset the token origin audience
jwt = self._introspect(access_token)
claims = self._verify_claims(jwt)
user_info = self._verify_user_info(claims["token_introspection"])
sub = user_info.get("sub")
if sub is None:
message = "User info contained no recognizable user identification"
logger.debug(message)
raise SuspiciousOperation(message)
try:
user = self.UserModel.objects.get(sub=sub)
except self.UserModel.DoesNotExist:
logger.debug("Login failed: No user with %s found", sub)
return None
self.token_origin_audience = str(user_info["aud"])
return user
def _verify_user_info(self, introspection_response):
"""Verify the 'introspection_response' to get valid and relevant user info.
The 'introspection_response' should be still active, and while authenticating
the resource owner should have requested relevant scope to access her data in
our resource server.
Scope should be configured to match between the AS and the RS. The AS will filter
all the scopes the resource owner requested to expose only the relevant ones to
our resource server.
"""
active = introspection_response.get("active", None)
if not active:
message = "Introspection response is not active."
logger.debug(message)
raise SuspiciousOperation(message)
requested_scopes = introspection_response.get("scope", None).split(" ")
if set(self._scopes).isdisjoint(set(requested_scopes)):
message = "Introspection response contains any required scopes."
logger.debug(message)
raise SuspiciousOperation(message)
audience = introspection_response.get("aud", None)
if not audience:
raise SuspiciousOperation(
"Introspection response does not provide source audience."
)
return introspection_response
def _introspect(self, token):
"""Introspect an access token to the authorization server."""
try:
jwe = self._authorization_server_client.get_introspection(
self._client_id,
self._client_secret,
token,
)
except HTTPError as err:
message = "Could not fetch introspection"
logger.debug("%s. Exception:", message, exc_info=True)
raise SuspiciousOperation(message) from err
private_key = utils.import_private_key_from_settings()
jws = self._decrypt(jwe, private_key=private_key)
try:
public_key_set = self._authorization_server_client.import_public_keys()
except (TypeError, ValueError, AttributeError, HTTPError) as err:
message = "Could get authorization server JWKS"
logger.debug("%s. Exception:", message, exc_info=True)
raise SuspiciousOperation(message) from err
jwt = self._decode(jws, public_key_set)
return jwt
def _decrypt(self, encrypted_token, private_key):
"""Decrypt the token encrypted by the Authorization Server (AS).
Resource Server (RS)'s public key is used for encryption, and its private
key is used for decryption. The RS's public key is exposed to the AS via a JWKS endpoint.
Encryption Algorithm and Encoding should be configured to match between the AS
and the RS.
"""
try:
decrypted_token = jose_jwe.decrypt_compact(
encrypted_token,
private_key,
algorithms=[self._encryption_algorithm, self._encryption_encoding],
)
except Exception as err:
message = "Token decryption failed"
logger.debug("%s. Exception:", message, exc_info=True)
raise SuspiciousOperation(message) from err
return decrypted_token
def _decode(self, encoded_token, public_key_set):
"""Decode the token signed by the Authorization Server (AS).
AS's private key is used for signing, and its public key is used for decoding.
The AS public key is exposed via a JWK endpoint.
Signing Algorithm should be configured to match between the AS and the RS.
"""
try:
token = jose_jwt.decode(
encoded_token.plaintext,
public_key_set,
algorithms=[self._signing_algorithm],
)
except ValueError as err:
message = "Token decoding failed"
logger.debug("%s. Exception:", message, exc_info=True)
raise SuspiciousOperation(message) from err
return token
def _verify_claims(self, token):
"""Verify the claims of the token to ensure authentication security.
By verifying these claims, we ensure that the token was issued by a
trusted authorization server and is intended for this specific
resource server. This prevents various types of attacks, such as
token substitution or misuse of tokens issued for different clients.
"""
try:
self._claims_registry.validate(token.claims)
except (InvalidClaimError, InvalidTokenError) as err:
message = "Failed to validate token's claims"
logger.debug("%s. Exception:", message, exc_info=True)
raise SuspiciousOperation(message) from err
return token.claims
class ResourceServerImproperlyConfiguredBackend:
"""Fallback backend for improperly configured Resource Servers."""
token_origin_audience = None
def get_or_create_user(self, access_token, id_token, payload):
"""Indicate that the Resource Server is improperly configured."""
raise AuthenticationFailed("Resource Server is improperly configured")
@@ -1,97 +0,0 @@
"""Resource Server Clients classes"""
from django.core.exceptions import ImproperlyConfigured
import requests
from joserfc.jwk import KeySet
class AuthorizationServerClient:
"""Client for interacting with an OAuth 2.0 authorization server.
An authorization server issues access tokens to client applications after authenticating
and obtaining authorization from the resource owner. It also provides endpoints for token
introspection and JSON Web Key Sets (JWKS) to validate and decode tokens.
This client facilitates communication with the authorization server, including:
- Fetching token introspection responses.
- Fetching JSON Web Key Sets (JWKS) for token validation.
- Setting appropriate headers for secure communication as recommended by RFC drafts.
"""
# ruff: noqa: PLR0913 PLR0917
# pylint: disable=too-many-positional-arguments
# pylint: disable=too-many-arguments
def __init__(
self,
url,
url_jwks,
url_introspection,
verify_ssl,
timeout,
proxy,
):
"""Require at a minimum url, url_jwks and url_introspection."""
if not url or not url_jwks or not url_introspection:
raise ImproperlyConfigured(
"Could not instantiate AuthorizationServerClient, some parameters are missing."
)
self.url = url
self._url_introspection = url_introspection
self._url_jwks = url_jwks
self._verify_ssl = verify_ssl
self._timeout = timeout
self._proxy = proxy
@property
def _introspection_headers(self):
"""Get HTTP header for the introspection request.
Notify the authorization server that we expect a signed and encrypted response
by setting the appropriate 'Accept' header.
This follows the recommendation from the draft RFC:
https://datatracker.ietf.org/doc/html/draft-ietf-oauth-jwt-introspection-response-12.
"""
return {
"Content-Type": "application/x-www-form-urlencoded",
"Accept": "application/token-introspection+jwt",
}
def get_introspection(self, client_id, client_secret, token):
"""Retrieve introspection response about a token."""
response = requests.post(
self._url_introspection,
data={
"client_id": client_id,
"client_secret": client_secret,
"token": token,
},
headers=self._introspection_headers,
verify=self._verify_ssl,
timeout=self._timeout,
proxies=self._proxy,
)
response.raise_for_status()
return response.text
def get_jwks(self):
"""Retrieve Authorization Server JWKS."""
response = requests.get(
self._url_jwks,
verify=self._verify_ssl,
timeout=self._timeout,
proxies=self._proxy,
)
response.raise_for_status()
return response.json()
def import_public_keys(self):
"""Retrieve and import Authorization Server JWKS."""
jwks = self.get_jwks()
public_keys = KeySet.import_key_set(jwks)
return public_keys
@@ -1,53 +0,0 @@
"""
Mixins for resource server views.
"""
from rest_framework import exceptions as drf_exceptions
from .authentication import ResourceServerAuthentication
class ResourceServerMixin:
"""
Mixin for resource server views:
- Restrict the authentication class to ResourceServerAuthentication.
- Adds the Service Provider ID to the serializer context.
- Fetch the Service Provider ID from the OIDC introspected token stored
in the request.
This Mixin *must* be used in every view that should act as a resource server.
"""
authentication_classes = [ResourceServerAuthentication]
def get_serializer_context(self):
"""Extra context provided to the serializer class."""
context = super().get_serializer_context()
# When used as a resource server, we need to know the audience to automatically:
# - add the Service Provider to the team "scope" on creation
context["from_service_provider_audience"] = (
self._get_service_provider_audience()
)
return context
def _get_service_provider_audience(self):
"""Return the audience of the Service Provider from the OIDC introspected token."""
if not isinstance(
self.request.successful_authenticator, ResourceServerAuthentication
):
# We could check request.resource_server_token_audience here, but it's
# more explicit to check the authenticator type and assert the attribute
# existence.
return None
# When used as a resource server, the request has a token audience
service_provider_audience = self.request.resource_server_token_audience
if not service_provider_audience: # should not happen
raise drf_exceptions.AuthenticationFailed(
"Resource server token audience not found in request"
)
return service_provider_audience
-9
View File
@@ -1,9 +0,0 @@
"""Resource Server URL Configuration"""
from django.urls import path
from .views import JWKSView
urlpatterns = [
path("jwks", JWKSView.as_view(), name="resource_server_jwks"),
]
-48
View File
@@ -1,48 +0,0 @@
"""Resource Server utils functions"""
from django.conf import settings
from django.core.exceptions import ImproperlyConfigured
from joserfc.jwk import JWKRegistry
def import_private_key_from_settings():
"""Import the private key used by the resource server when interacting with the OIDC provider.
This private key is crucial; its public components are exposed in the JWK endpoints,
while its private component is used for decrypting the introspection token retrieved
from the OIDC provider.
By default, we recommend using RSAKey for asymmetric encryption,
known for its strong security features.
Note:
- The function requires the 'OIDC_RS_PRIVATE_KEY_STR' setting to be configured.
- The 'OIDC_RS_ENCRYPTION_KEY_TYPE' and 'OIDC_RS_ENCRYPTION_ALGO' settings can be customized
based on the chosen key type.
Raises:
ImproperlyConfigured: If the private key setting is missing, empty, or incorrect.
Returns:
joserfc.jwk.JWK: The imported private key as a JWK object.
"""
private_key_str = getattr(settings, "OIDC_RS_PRIVATE_KEY_STR", None)
if not private_key_str:
raise ImproperlyConfigured(
"OIDC_RS_PRIVATE_KEY_STR setting is missing or empty."
)
private_key_pem = private_key_str.encode()
try:
private_key = JWKRegistry.import_key(
private_key_pem,
key_type=settings.OIDC_RS_ENCRYPTION_KEY_TYPE,
parameters={"alg": settings.OIDC_RS_ENCRYPTION_ALGO, "use": "enc"},
)
except ValueError as err:
raise ImproperlyConfigured("OIDC_RS_PRIVATE_KEY_STR setting is wrong.") from err
return private_key
-40
View File
@@ -1,40 +0,0 @@
"""Resource Server views"""
from django.core.exceptions import ImproperlyConfigured
from joserfc.jwk import KeySet
from rest_framework.response import Response
from rest_framework.views import APIView
from . import utils
class JWKSView(APIView):
"""
API endpoint for retrieving a JSON Web Keys Set (JWKS).
Returns:
Response: JSON response containing the JWKS data.
"""
authentication_classes = [] # disable authentication
permission_classes = [] # disable permission
def get(self, request):
"""Handle GET requests to retrieve JSON Web Keys Set (JWKS).
Returns:
Response: JSON response containing the JWKS data.
"""
try:
private_key = utils.import_private_key_from_settings()
except (ImproperlyConfigured, ValueError) as err:
return Response({"error": str(err)}, status=500)
try:
jwk = KeySet([private_key]).as_dict(private=False)
except (TypeError, ValueError, AttributeError):
return Response({"error": "Could not load key"}, status=500)
return Response(jwk)
Binary file not shown.

Before

Width:  |  Height:  |  Size: 13 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 5.5 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 40 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 1.8 KiB

@@ -1 +0,0 @@
"""Core template tags module."""
@@ -1,58 +0,0 @@
"""Custom template tags for the core application of People."""
import base64
from django import template
from django.contrib.staticfiles import finders
from PIL import ImageFile as PillowImageFile
register = template.Library()
def image_to_base64(file_or_path, close=False):
"""
Return the src string of the base64 encoding of an image represented by its path
or file opened or not.
Inspired by Django's "get_image_dimensions"
"""
pil_parser = PillowImageFile.Parser()
if hasattr(file_or_path, "read"):
file = file_or_path
if file.closed and hasattr(file, "open"):
file_or_path.open()
file_pos = file.tell()
file.seek(0)
else:
try:
# pylint: disable=consider-using-with
file = open(file_or_path, "rb")
except OSError:
return ""
close = True
try:
image_data = file.read()
if not image_data:
return ""
pil_parser.feed(image_data)
if pil_parser.image:
mime_type = pil_parser.image.get_format_mimetype()
encoded_string = base64.b64encode(image_data)
return f"data:{mime_type:s};base64, {encoded_string.decode('utf-8'):s}"
return ""
finally:
if close:
file.close()
else:
file.seek(file_pos)
@register.simple_tag
def base64_static(path):
"""Return a static file into a base64."""
full_path = finders.find(path)
if full_path:
return image_to_base64(full_path, True)
return ""
-1
View File
@@ -1 +0,0 @@
"""Core tests."""
@@ -1 +0,0 @@
"""Core authentication tests."""
@@ -1,455 +0,0 @@
"""Unit tests for the Authentication Backends."""
from django.contrib.auth import get_user_model
from django.core.exceptions import SuspiciousOperation
import pytest
from core import factories, models
from core.authentication.backends import OIDCAuthenticationBackend
pytestmark = pytest.mark.django_db
User = get_user_model()
def test_authentication_getter_existing_user_no_email(
django_assert_num_queries, monkeypatch
):
"""
If an existing user matches the user's info sub, the user should be returned.
"""
klass = OIDCAuthenticationBackend()
user = factories.UserFactory()
def get_userinfo_mocked(*args):
return {"sub": user.sub}
monkeypatch.setattr(OIDCAuthenticationBackend, "get_userinfo", get_userinfo_mocked)
with django_assert_num_queries(1):
authenticated_user = klass.get_or_create_user(
access_token="test-token", id_token=None, payload=None
)
assert user == authenticated_user
def test_authentication_getter_existing_user_with_email(
django_assert_num_queries, monkeypatch
):
"""
When the user's info contains an email and targets an existing user,
"""
klass = OIDCAuthenticationBackend()
user = factories.UserFactory(name="John Doe", with_organization=True)
def get_userinfo_mocked(*args):
return {
"sub": user.sub,
"email": user.email,
"first_name": "John",
"last_name": "Doe",
}
monkeypatch.setattr(OIDCAuthenticationBackend, "get_userinfo", get_userinfo_mocked)
# Only 1 query because email and names have not changed
with django_assert_num_queries(1):
authenticated_user = klass.get_or_create_user(
access_token="test-token", id_token=None, payload=None
)
assert user == authenticated_user
@pytest.mark.parametrize(
"first_name, last_name, email",
[
("Jack", "Doe", "john.doe@example.com"),
("John", "Duy", "john.doe@example.com"),
("John", "Doe", "jack.duy@example.com"),
("Jack", "Duy", "jack.duy@example.com"),
],
)
def test_authentication_getter_existing_user_change_fields(
first_name, last_name, email, django_assert_num_queries, monkeypatch
):
"""
It should update the email or name fields on the user when they change.
"""
klass = OIDCAuthenticationBackend()
user = factories.UserFactory(
name="John Doe", email="john.doe@example.com", with_organization=True
)
def get_userinfo_mocked(*args):
return {
"sub": user.sub,
"email": email,
"first_name": first_name,
"last_name": last_name,
}
monkeypatch.setattr(OIDCAuthenticationBackend, "get_userinfo", get_userinfo_mocked)
# One and only one additional update query when a field has changed
with django_assert_num_queries(2):
authenticated_user = klass.get_or_create_user(
access_token="test-token", id_token=None, payload=None
)
assert user == authenticated_user
user.refresh_from_db()
assert user.email == email
assert user.name == f"{first_name:s} {last_name:s}"
def test_authentication_getter_existing_user_keep_fields(
django_assert_num_queries, monkeypatch
):
"""
Falsy values in claim should not update the user's fields.
"""
klass = OIDCAuthenticationBackend()
user = factories.UserFactory(
name="John Doe", email="john.doe@example.com", with_organization=True
)
def get_userinfo_mocked(*args):
return {
"sub": user.sub,
"email": None,
"first_name": "",
"last_name": "",
}
monkeypatch.setattr(OIDCAuthenticationBackend, "get_userinfo", get_userinfo_mocked)
# No field changed no more query
with django_assert_num_queries(1):
authenticated_user = klass.get_or_create_user(
access_token="test-token", id_token=None, payload=None
)
assert user == authenticated_user
user.refresh_from_db()
assert user.email == "john.doe@example.com"
assert user.name == "John Doe"
def test_authentication_getter_existing_user_via_email(
django_assert_num_queries, monkeypatch
):
"""
If an existing user doesn't match the sub but matches the email,
the user should be returned.
"""
klass = OIDCAuthenticationBackend()
db_user = factories.UserFactory(with_organization=True)
def get_userinfo_mocked(*args):
return {"sub": "123", "email": db_user.email}
monkeypatch.setattr(OIDCAuthenticationBackend, "get_userinfo", get_userinfo_mocked)
with django_assert_num_queries(2):
user = klass.get_or_create_user(
access_token="test-token", id_token=None, payload=None
)
assert user == db_user
def test_authentication_getter_existing_user_no_fallback_to_email(
settings, monkeypatch
):
"""
When the "OIDC_FALLBACK_TO_EMAIL_FOR_IDENTIFICATION" setting is set to False,
the system should not match users by email, even if the email matches.
"""
klass = OIDCAuthenticationBackend()
db_user = factories.UserFactory()
# Set the setting to False
settings.OIDC_FALLBACK_TO_EMAIL_FOR_IDENTIFICATION = False
def get_userinfo_mocked(*args):
return {"sub": "123", "email": db_user.email}
monkeypatch.setattr(OIDCAuthenticationBackend, "get_userinfo", get_userinfo_mocked)
user = klass.get_or_create_user(
access_token="test-token", id_token=None, payload=None
)
# Since the sub doesn't match, it should create a new user
assert models.User.objects.count() == 2
assert user != db_user
assert user.sub == "123"
def test_authentication_getter_new_user_no_email(monkeypatch):
"""
If no user matches the user's info sub, a user should not be created without email
nor organization registration ID.
"""
klass = OIDCAuthenticationBackend()
def get_userinfo_mocked(*args):
return {"sub": "123"} # No email, no organization registration ID
monkeypatch.setattr(OIDCAuthenticationBackend, "get_userinfo", get_userinfo_mocked)
with (
pytest.raises(
SuspiciousOperation,
match="Claims contained no recognizable organization identification",
),
):
klass.get_or_create_user(access_token="test-token", id_token=None, payload=None)
def test_authentication_getter_new_user_with_email(monkeypatch):
"""
If no user matches the user's info sub, a user should be created.
User's email and name should be set on the user.
"""
klass = OIDCAuthenticationBackend()
email = "people@example.com"
def get_userinfo_mocked(*args):
return {"sub": "123", "email": email, "first_name": "John", "last_name": "Doe"}
monkeypatch.setattr(OIDCAuthenticationBackend, "get_userinfo", get_userinfo_mocked)
user = klass.get_or_create_user(
access_token="test-token", id_token=None, payload=None
)
assert user.sub == "123"
assert user.email == email
assert user.name == "John Doe"
assert user.password == "!"
assert models.User.objects.count() == 1
def test_models_oidc_user_getter_invalid_token(django_assert_num_queries, monkeypatch):
"""The user's info doesn't contain a sub."""
klass = OIDCAuthenticationBackend()
def get_userinfo_mocked(*args):
return {
"test": "123",
}
monkeypatch.setattr(OIDCAuthenticationBackend, "get_userinfo", get_userinfo_mocked)
with (
django_assert_num_queries(0),
pytest.raises(
SuspiciousOperation,
match="User info contained no recognizable user identification",
),
):
klass.get_or_create_user(access_token="test-token", id_token=None, payload=None)
assert models.User.objects.exists() is False
def test_authentication_getter_existing_disabled_user_via_sub(
django_assert_num_queries, monkeypatch
):
"""
If an existing user matches the sub but is disabled,
an error should be raised and a user should not be created.
"""
klass = OIDCAuthenticationBackend()
db_user = factories.UserFactory(name="John Doe", is_active=False)
def get_userinfo_mocked(*args):
return {
"sub": db_user.sub,
"email": db_user.email,
"first_name": "John",
"last_name": "Doe",
}
monkeypatch.setattr(OIDCAuthenticationBackend, "get_userinfo", get_userinfo_mocked)
with (
django_assert_num_queries(1),
pytest.raises(SuspiciousOperation, match="User account is disabled"),
):
klass.get_or_create_user(access_token="test-token", id_token=None, payload=None)
assert models.User.objects.count() == 1
def test_authentication_getter_existing_disabled_user_via_email(
django_assert_num_queries, monkeypatch
):
"""
If an existing user does not matches the sub but match the email and is disabled,
an error should be raised and a user should not be created.
"""
klass = OIDCAuthenticationBackend()
db_user = factories.UserFactory(name="John Doe", is_active=False)
def get_userinfo_mocked(*args):
return {
"sub": "random",
"email": db_user.email,
"first_name": "John",
"last_name": "Doe",
}
monkeypatch.setattr(OIDCAuthenticationBackend, "get_userinfo", get_userinfo_mocked)
with (
django_assert_num_queries(2),
pytest.raises(SuspiciousOperation, match="User account is disabled"),
):
klass.get_or_create_user(access_token="test-token", id_token=None, payload=None)
assert models.User.objects.count() == 1
def test_authentication_getter_new_user_with_email_new_organization(monkeypatch):
"""
If no user matches the user's info sub, a user should be created.
If the corresponding organization doesn't exist, it should be created.
"""
klass = OIDCAuthenticationBackend()
email = "people@example.com"
def get_userinfo_mocked(*args):
return {"sub": "123", "email": email, "first_name": "John", "last_name": "Doe"}
monkeypatch.setattr(OIDCAuthenticationBackend, "get_userinfo", get_userinfo_mocked)
user = klass.get_or_create_user(
access_token="test-token", id_token=None, payload=None
)
assert user.organization is not None
assert user.organization.domain_list == ["example.com"]
assert user.organization.registration_id_list == []
@pytest.mark.parametrize(
"registration_id_setting,expected_registration_id_list,expected_domain_list",
[
(None, [], ["example.com"]),
("missing-claim", [], ["example.com"]),
("registration_number", ["12345678901234"], []),
],
)
def test_authentication_getter_new_user_with_registration_id_new_organization(
monkeypatch,
settings,
registration_id_setting,
expected_registration_id_list,
expected_domain_list,
):
"""
If no user matches the user's info sub, a user should be created.
If the corresponding organization doesn't exist, it should be created.
"""
settings.OIDC_ORGANIZATION_REGISTRATION_ID_FIELD = registration_id_setting
klass = OIDCAuthenticationBackend()
email = "people@example.com"
def get_userinfo_mocked(*args):
return {
"sub": "123",
"email": email,
"first_name": "John",
"last_name": "Doe",
"registration_number": "12345678901234",
}
monkeypatch.setattr(OIDCAuthenticationBackend, "get_userinfo", get_userinfo_mocked)
user = klass.get_or_create_user(
access_token="test-token", id_token=None, payload=None
)
assert user.organization is not None
assert user.organization.domain_list == expected_domain_list
assert user.organization.registration_id_list == expected_registration_id_list
def test_authentication_getter_existing_user_via_email_update_organization(
django_assert_num_queries, monkeypatch
):
"""
If an existing user already exists without organization, the organization must be updated.
"""
klass = OIDCAuthenticationBackend()
db_user = factories.UserFactory(name="John Doe", email="toto@my-domain.com")
def get_userinfo_mocked(*args):
return {
"sub": db_user.sub,
"email": db_user.email,
"first_name": "John",
"last_name": "Doe",
}
monkeypatch.setattr(OIDCAuthenticationBackend, "get_userinfo", get_userinfo_mocked)
with django_assert_num_queries(9):
user = klass.get_or_create_user(
access_token="test-token", id_token=None, payload=None
)
assert user == db_user
assert user.organization is not None
assert user.organization.domain_list == ["my-domain.com"]
def test_authentication_getter_existing_user_with_registration_id(
monkeypatch,
settings,
):
"""
Authenticate a user who already exists with an organization registration ID.
This asserts the "update_user_if_needed" does not fail when the claim
contains the organization registration ID (or any value not present in the
User model).
"""
settings.OIDC_ORGANIZATION_REGISTRATION_ID_FIELD = "registration_number"
klass = OIDCAuthenticationBackend()
_existing_user = factories.UserFactory(
sub="123",
name="John Doe",
email="people@example.com",
)
def get_userinfo_mocked(*args):
return {
"sub": "123",
"email": "people@example.com",
"first_name": "John",
"last_name": "Doe",
"registration_number": "12345678901234",
}
monkeypatch.setattr(OIDCAuthenticationBackend, "get_userinfo", get_userinfo_mocked)
user = klass.get_or_create_user(
access_token="test-token", id_token=None, payload=None
)
assert user.organization is not None
assert user.organization.registration_id_list == ["12345678901234"]
@@ -1,10 +0,0 @@
"""Unit tests for the Authentication URLs."""
from core.authentication.urls import urlpatterns
def test_urls_override_default_mozilla_django_oidc():
"""Custom URL patterns should override default ones from Mozilla Django OIDC."""
url_names = [u.name for u in urlpatterns]
assert url_names.index("oidc_logout_custom") < url_names.index("oidc_logout")
@@ -1,231 +0,0 @@
"""Unit tests for the Authentication Views."""
from unittest import mock
from urllib.parse import parse_qs, urlparse
from django.contrib.auth.models import AnonymousUser
from django.contrib.sessions.middleware import SessionMiddleware
from django.core.exceptions import SuspiciousOperation
from django.test import RequestFactory
from django.test.utils import override_settings
from django.urls import reverse
from django.utils import crypto
import pytest
from rest_framework.test import APIClient
from core import factories
from core.authentication.views import OIDCLogoutCallbackView, OIDCLogoutView
pytestmark = pytest.mark.django_db
@override_settings(LOGOUT_REDIRECT_URL="/example-logout")
def test_view_logout_anonymous():
"""Anonymous users calling the logout url,
should be redirected to the specified LOGOUT_REDIRECT_URL."""
url = reverse("oidc_logout_custom")
response = APIClient().get(url)
assert response.status_code == 302
assert response.url == "/example-logout"
@mock.patch.object(
OIDCLogoutView, "construct_oidc_logout_url", return_value="/example-logout"
)
def test_view_logout(mocked_oidc_logout_url):
"""Authenticated users should be redirected to OIDC provider for logout."""
user = factories.UserFactory()
client = APIClient()
client.force_login(user)
url = reverse("oidc_logout_custom")
response = client.get(url)
mocked_oidc_logout_url.assert_called_once()
assert response.status_code == 302
assert response.url == "/example-logout"
@override_settings(LOGOUT_REDIRECT_URL="/default-redirect-logout")
@mock.patch.object(
OIDCLogoutView, "construct_oidc_logout_url", return_value="/default-redirect-logout"
)
def test_view_logout_no_oidc_provider(mocked_oidc_logout_url):
"""Authenticated users should be logged out when no OIDC provider is available."""
user = factories.UserFactory()
client = APIClient()
client.force_login(user)
url = reverse("oidc_logout_custom")
with mock.patch("mozilla_django_oidc.views.auth.logout") as mock_logout:
response = client.get(url)
mocked_oidc_logout_url.assert_called_once()
mock_logout.assert_called_once()
assert response.status_code == 302
assert response.url == "/default-redirect-logout"
@override_settings(LOGOUT_REDIRECT_URL="/example-logout")
def test_view_logout_callback_anonymous():
"""Anonymous users calling the logout callback url,
should be redirected to the specified LOGOUT_REDIRECT_URL."""
url = reverse("oidc_logout_callback")
response = APIClient().get(url)
assert response.status_code == 302
assert response.url == "/example-logout"
@pytest.mark.parametrize(
"initial_oidc_states",
[{}, {"other_state": "foo"}],
)
def test_view_logout_persist_state(initial_oidc_states):
"""State value should be persisted in session's data."""
user = factories.UserFactory()
request = RequestFactory().request()
request.user = user
middleware = SessionMiddleware(get_response=lambda x: x)
middleware.process_request(request)
if initial_oidc_states:
request.session["oidc_states"] = initial_oidc_states
request.session.save()
mocked_state = "mock_state"
OIDCLogoutView().persist_state(request, mocked_state)
assert "oidc_states" in request.session
assert request.session["oidc_states"] == {
"mock_state": {},
**initial_oidc_states,
}
@override_settings(OIDC_OP_LOGOUT_ENDPOINT="/example-logout")
@mock.patch.object(OIDCLogoutView, "persist_state")
@mock.patch.object(crypto, "get_random_string", return_value="mocked_state")
def test_view_logout_construct_oidc_logout_url(
mocked_get_random_string, mocked_persist_state
):
"""Should construct the logout URL to initiate the logout flow with the OIDC provider."""
user = factories.UserFactory()
request = RequestFactory().request()
request.user = user
middleware = SessionMiddleware(get_response=lambda x: x)
middleware.process_request(request)
request.session["oidc_id_token"] = "mocked_oidc_id_token"
request.session.save()
redirect_url = OIDCLogoutView().construct_oidc_logout_url(request)
mocked_persist_state.assert_called_once()
mocked_get_random_string.assert_called_once()
params = parse_qs(urlparse(redirect_url).query)
assert params["id_token_hint"][0] == "mocked_oidc_id_token"
assert params["state"][0] == "mocked_state"
url = reverse("oidc_logout_callback")
assert url in params["post_logout_redirect_uri"][0]
@override_settings(LOGOUT_REDIRECT_URL="/")
def test_view_logout_construct_oidc_logout_url_none_id_token():
"""If no ID token is available in the session,
the user should be redirected to the final URL."""
user = factories.UserFactory()
request = RequestFactory().request()
request.user = user
middleware = SessionMiddleware(get_response=lambda x: x)
middleware.process_request(request)
redirect_url = OIDCLogoutView().construct_oidc_logout_url(request)
assert redirect_url == "/"
@pytest.mark.parametrize(
"initial_state",
[None, {"other_state": "foo"}],
)
def test_view_logout_callback_wrong_state(initial_state):
"""Should raise an error if OIDC state doesn't match session data."""
user = factories.UserFactory()
request = RequestFactory().request()
request.user = user
middleware = SessionMiddleware(get_response=lambda x: x)
middleware.process_request(request)
if initial_state:
request.session["oidc_states"] = initial_state
request.session.save()
callback_view = OIDCLogoutCallbackView.as_view()
with pytest.raises(SuspiciousOperation) as excinfo:
callback_view(request)
assert (
str(excinfo.value) == "OIDC callback state not found in session `oidc_states`!"
)
@override_settings(LOGOUT_REDIRECT_URL="/example-logout")
def test_view_logout_callback():
"""If state matches, callback should clear OIDC state and redirects."""
user = factories.UserFactory()
request = RequestFactory().get("/logout-callback/", data={"state": "mocked_state"})
request.user = user
middleware = SessionMiddleware(get_response=lambda x: x)
middleware.process_request(request)
mocked_state = "mocked_state"
request.session["oidc_states"] = {mocked_state: {}}
request.session.save()
callback_view = OIDCLogoutCallbackView.as_view()
with mock.patch("mozilla_django_oidc.views.auth.logout") as mock_logout:
def clear_user(request):
# Assert state is cleared prior to logout
assert request.session["oidc_states"] == {}
request.user = AnonymousUser()
mock_logout.side_effect = clear_user
response = callback_view(request)
mock_logout.assert_called_once()
assert response.status_code == 302
assert response.url == "/example-logout"
@@ -1 +0,0 @@
"""Contacts tests package."""
@@ -1,213 +0,0 @@
"""
Test contacts API endpoints in People's core app.
"""
from django.test.utils import override_settings
import pytest
from rest_framework.test import APIClient
from core import factories, models
pytestmark = pytest.mark.django_db
CONTACT_DATA = {
"emails": [
{"type": "Work", "value": "john.doe@work.com"},
{"type": "Home", "value": "john.doe@home.com"},
],
"phones": [
{"type": "Work", "value": "(123) 456-7890"},
{"type": "Other", "value": "(987) 654-3210"},
],
"addresses": [
{
"type": "Home",
"street": "123 Main St",
"city": "Cityville",
"state": "CA",
"zip": "12345",
"country": "USA",
}
],
"links": [
{"type": "Blog", "value": "http://personalwebsite.com"},
{"type": "Website", "value": "http://workwebsite.com"},
],
"customFields": {"custom_field_1": "value1", "custom_field_2": "value2"},
"organizations": [
{
"name": "ACME Corporation",
"department": "IT",
"jobTitle": "Software Engineer",
},
{
"name": "XYZ Ltd",
"department": "Marketing",
"jobTitle": "Marketing Specialist",
},
],
}
def test_api_contacts_create_anonymous_forbidden():
"""Anonymous users should not be able to create contacts via the API."""
response = APIClient().post(
"/api/v1.0/contacts/",
{
"full_name": "David",
"short_name": "Bowman",
},
)
assert response.status_code == 401
assert not models.Contact.objects.exists()
def test_api_contacts_create_authenticated_missing_base():
"""Authenticated user should be able to create contact without override."""
user = factories.UserFactory()
client = APIClient()
client.force_login(user)
response = client.post(
"/api/v1.0/contacts/",
{
"full_name": "David Bowman",
"short_name": "Dave",
"data": {},
},
format="json",
)
assert response.status_code == 201
new_contact = models.Contact.objects.get(owner=user)
assert response.json() == {
"override": None,
"abilities": {"delete": True, "get": True, "patch": True, "put": True},
"data": {},
"full_name": "David Bowman",
"id": str(new_contact.pk),
"notes": "",
"owner": str(user.pk),
"short_name": "Dave",
}
def test_api_contacts_create_authenticated_successful():
"""Authenticated users should be able to create contacts."""
user = factories.UserFactory()
base_contact = factories.BaseContactFactory()
client = APIClient()
client.force_login(user)
# Existing override for another user should not interfere
factories.ContactFactory(override=base_contact)
response = client.post(
"/api/v1.0/contacts/",
{
"override": str(base_contact.id),
"full_name": "David Bowman",
"short_name": "Dave",
"data": CONTACT_DATA,
},
format="json",
)
assert response.status_code == 201
assert models.Contact.objects.count() == 3
contact = models.Contact.objects.get(owner=user)
assert response.json() == {
"id": str(contact.id),
"abilities": {"delete": True, "get": True, "patch": True, "put": True},
"override": str(base_contact.id),
"data": CONTACT_DATA,
"full_name": "David Bowman",
"notes": "",
"owner": str(user.id),
"short_name": "Dave",
}
assert contact.full_name == "David Bowman"
assert contact.short_name == "Dave"
assert contact.data == CONTACT_DATA
assert contact.override == base_contact
assert contact.owner == user
@override_settings(ALLOW_API_USER_CREATE=True)
def test_api_contacts_create_authenticated_existing_override():
"""
Trying to create a contact overriding a contact that is already overridden by the user
should receive a 400 error.
"""
user = factories.UserFactory()
base_contact = factories.BaseContactFactory()
factories.ContactFactory(override=base_contact, owner=user)
client = APIClient()
client.force_login(user)
response = client.post(
"/api/v1.0/contacts/",
{
"override": str(base_contact.id),
"full_name": "David Bowman",
"notes": "",
"short_name": "Dave",
"data": CONTACT_DATA,
},
format="json",
)
assert response.status_code == 400
assert models.Contact.objects.count() == 2
assert response.json() == {
"__all__": ["Contact with this Owner and Override already exists."]
}
def test_api_contacts_create_authenticated_successful_with_notes():
"""Authenticated users should be able to create contacts with notes."""
user = factories.UserFactory()
client = APIClient()
client.force_login(user)
response = client.post(
"/api/v1.0/contacts/",
{
"full_name": "David Bowman",
"short_name": "Dave",
"data": CONTACT_DATA,
"notes": "This is a note",
},
format="json",
)
assert response.status_code == 201
assert models.Contact.objects.count() == 1
contact = models.Contact.objects.get(owner=user)
assert response.json() == {
"id": str(contact.id),
"abilities": {"delete": True, "get": True, "patch": True, "put": True},
"override": None,
"data": CONTACT_DATA,
"full_name": "David Bowman",
"notes": "This is a note",
"owner": str(user.id),
"short_name": "Dave",
}
assert contact.full_name == "David Bowman"
assert contact.short_name == "Dave"
assert contact.data == CONTACT_DATA
assert contact.owner == user
assert contact.notes == "This is a note"
@@ -1,120 +0,0 @@
"""
Test contacts API endpoints in People's core app.
"""
import pytest
from rest_framework.test import APIClient
from core import factories, models
pytestmark = pytest.mark.django_db
def test_api_contacts_delete_list_anonymous():
"""Anonymous users should not be allowed to delete a list of contacts."""
factories.ContactFactory.create_batch(2)
response = APIClient().delete("/api/v1.0/contacts/")
assert response.status_code == 401
assert models.Contact.objects.count() == 2
def test_api_contacts_delete_list_authenticated():
"""Authenticated users should not be allowed to delete a list of contacts."""
user = factories.UserFactory()
client = APIClient()
client.force_login(user)
factories.ContactFactory.create_batch(2)
response = client.delete("/api/v1.0/contacts/")
assert response.status_code == 405
assert models.Contact.objects.count() == 2
def test_api_contacts_delete_anonymous():
"""Anonymous users should not be allowed to delete a contact."""
contact = factories.ContactFactory()
client = APIClient()
response = client.delete(f"/api/v1.0/contacts/{contact.id!s}/")
assert response.status_code == 401
assert models.Contact.objects.count() == 1
def test_api_contacts_delete_authenticated_public():
"""
Authenticated users should not be allowed to delete a public contact.
"""
user = factories.UserFactory()
client = APIClient()
client.force_login(user)
contact = factories.BaseContactFactory()
response = client.delete(
f"/api/v1.0/contacts/{contact.id!s}/",
)
assert response.status_code == 403
assert models.Contact.objects.count() == 1
def test_api_contacts_delete_authenticated_owner():
"""
Authenticated users should be allowed to delete a contact they own.
"""
user = factories.UserFactory()
contact = factories.ContactFactory(owner=user)
client = APIClient()
client.force_login(user)
response = client.delete(
f"/api/v1.0/contacts/{contact.id!s}/",
)
assert response.status_code == 204
assert models.Contact.objects.count() == 0
assert models.Contact.objects.filter(id=contact.id).exists() is False
def test_api_contacts_delete_authenticated_profile():
"""
Authenticated users should not be allowed to delete their profile contact.
"""
user = factories.UserFactory()
contact = factories.ProfileContactFactory(user=user)
client = APIClient()
client.force_login(user)
response = client.delete(
f"/api/v1.0/contacts/{contact.id!s}/",
)
assert response.status_code == 403
assert models.Contact.objects.exists() is True
def test_api_contacts_delete_authenticated_other():
"""
Authenticated users should not be allowed to delete a contact they don't own.
"""
user = factories.UserFactory()
contact = factories.ContactFactory()
client = APIClient()
client.force_login(user)
response = client.delete(
f"/api/v1.0/contacts/{contact.id!s}/",
)
assert response.status_code == 403
assert models.Contact.objects.count() == 1
@@ -1,305 +0,0 @@
"""
Test contacts API endpoints in People's core app.
"""
import pytest
from rest_framework.test import APIClient
from core import factories
pytestmark = pytest.mark.django_db
def test_api_contacts_list_anonymous():
"""Anonymous users should not be allowed to list contacts."""
factories.ContactFactory.create_batch(2)
response = APIClient().get("/api/v1.0/contacts/")
assert response.status_code == 401
assert response.json() == {
"detail": "Authentication credentials were not provided."
}
def test_api_contacts_list_authenticated_no_query(django_assert_num_queries):
"""
Authenticated users should be able to list contacts without applying a query.
Profile and overridden contacts should be excluded.
"""
organization = factories.OrganizationFactory(with_registration_id=True)
user = factories.UserFactory(organization=organization)
# The user's profile contact should be listed (why not)
user_profile_contact = factories.ContactFactory(
owner=user, user=user, full_name="Dave Bowman"
)
# A contact that belongs to another user should not be listed
factories.ContactFactory()
# even if from the same organization
factories.ContactFactory(owner__organization=organization)
# A profile contact should not be listed if from another organization
factories.ProfileContactFactory()
# A profile contact for someone in the same organization should be listed
profile_contact = factories.ProfileContactFactory(
user__organization=organization, full_name="Frank Poole"
)
# An overridden contact should not be listed, but the override must be
overriden_contact = factories.ProfileContactFactory(user__organization=organization)
override_contact = factories.ContactFactory(
owner=user, override=overriden_contact, full_name="Nicole Foole"
)
client = APIClient()
client.force_login(user)
with django_assert_num_queries(2):
response = client.get("/api/v1.0/contacts/")
assert response.status_code == 200
assert response.json() == [
{
"id": str(user_profile_contact.pk),
"abilities": {"delete": False, "get": True, "patch": True, "put": True},
"override": None,
"owner": str(user.pk),
"data": user_profile_contact.data,
"full_name": user_profile_contact.full_name,
"notes": "",
"short_name": user_profile_contact.short_name,
},
{
"id": str(profile_contact.pk),
"abilities": {"delete": False, "get": True, "patch": False, "put": False},
"override": None,
"owner": str(profile_contact.user.pk),
"data": profile_contact.data,
"full_name": profile_contact.full_name,
"notes": "",
"short_name": profile_contact.short_name,
},
{
"id": str(override_contact.pk),
"abilities": {"delete": True, "get": True, "patch": True, "put": True},
"override": str(overriden_contact.pk),
"owner": str(user.pk),
"data": override_contact.data,
"full_name": override_contact.full_name,
"notes": "",
"short_name": override_contact.short_name,
},
]
def test_api_contacts_list_authenticated_by_full_name():
"""
Authenticated users should be able to search users with a case insensitive and
partial query on the full name.
"""
user = factories.UserFactory(name="Prudence Crandall")
dave = factories.BaseContactFactory(full_name="David Bowman", data={})
nicole = factories.BaseContactFactory(full_name="Nicole Foole", data={})
frank = factories.BaseContactFactory(full_name="Frank Poole", data={})
factories.BaseContactFactory(full_name="Heywood Floyd", data={})
# Full query should work
client = APIClient()
client.force_login(user)
response = client.get("/api/v1.0/contacts/?q=David%20Bowman")
assert response.status_code == 200
contact_ids = [contact["id"] for contact in response.json()]
assert contact_ids == [str(dave.id)]
# Partial query should work
response = client.get("/api/v1.0/contacts/?q=ank")
assert response.status_code == 200
contact_ids = [contact["id"] for contact in response.json()]
assert contact_ids == [str(frank.id)]
response = client.get("/api/v1.0/contacts/?q=ole")
assert response.status_code == 200
contact_ids = [contact["id"] for contact in response.json()]
assert contact_ids == [str(frank.id), str(nicole.id)]
response = client.get("/api/v1.0/contacts/?q=ool")
assert response.status_code == 200
contact_ids = [contact["id"] for contact in response.json()]
assert contact_ids == [str(frank.id), str(nicole.id)]
def test_api_contacts_list_authenticated_by_email():
"""
Authenticated users should be able to search users with a case insensitive and
partial query on the email.
"""
user = factories.UserFactory()
dave = factories.BaseContactFactory(
full_name="0", # don't match on full name but allow ordering
data={
"emails": [
{"type": "Home", "value": "dave@personal.com"},
{"type": "Work", "value": "david.bowman@example.com"},
],
},
)
nicole = factories.BaseContactFactory(
full_name="1", # don't match on full name but allow ordering
data={
"emails": [
{"type": "Work", "value": "nicole.foole@example.com"},
],
},
)
frank = factories.BaseContactFactory(
full_name="2", # don't match on full name but allow ordering
data={
"emails": [
{"type": "Home", "value": "francky@personal.com"},
{"type": "Work", "value": "franck.poole@example.com"},
],
},
)
factories.BaseContactFactory(
full_name="3", # don't match on full name but allow ordering
data={
"emails": [
{"type": "Work", "value": "heywood.floyd@example.com"},
],
},
)
# Full query should work
client = APIClient()
client.force_login(user)
response = client.get("/api/v1.0/contacts/?q=david.bowman@example.com")
assert response.status_code == 200
contact_ids = [contact["id"] for contact in response.json()]
assert contact_ids == [str(dave.pk)]
# Partial query should work
response = client.get("/api/v1.0/contacts/?q=anc")
assert response.status_code == 200
contact_ids = [contact["id"] for contact in response.json()]
assert contact_ids == [str(frank.pk)]
response = client.get("/api/v1.0/contacts/?q=olé") # accented
assert response.status_code == 200
contact_ids = [contact["id"] for contact in response.json()]
assert contact_ids == [str(nicole.pk), str(frank.pk)]
response = client.get("/api/v1.0/contacts/?q=oOl") # mixed case
assert response.status_code == 200
contact_ids = [contact["id"] for contact in response.json()]
assert contact_ids == [str(nicole.pk), str(frank.pk)]
def test_api_contacts_list_authenticated_uppercase_content():
"""Upper case content should be found by lower case query."""
user = factories.UserFactory()
dave = factories.BaseContactFactory(full_name="EEE", short_name="AAA")
# Unaccented full name
client = APIClient()
client.force_login(user)
response = client.get("/api/v1.0/contacts/?q=eee")
assert response.status_code == 200
contact_ids = [contact["id"] for contact in response.json()]
assert contact_ids == [str(dave.id)]
# Unaccented short name
response = client.get("/api/v1.0/contacts/?q=aaa")
assert response.status_code == 200
contact_ids = [contact["id"] for contact in response.json()]
assert contact_ids == [str(dave.id)]
def test_api_contacts_list_authenticated_capital_query():
"""Upper case query should find lower case content."""
user = factories.UserFactory()
dave = factories.BaseContactFactory(full_name="eee", short_name="aaa")
client = APIClient()
client.force_login(user)
# Unaccented full name
response = client.get("/api/v1.0/contacts/?q=EEE")
assert response.status_code == 200
contact_ids = [contact["id"] for contact in response.json()]
assert contact_ids == [str(dave.id)]
# Unaccented short name
response = client.get("/api/v1.0/contacts/?q=AAA")
assert response.status_code == 200
contact_ids = [contact["id"] for contact in response.json()]
assert contact_ids == [str(dave.id)]
def test_api_contacts_list_authenticated_accented_content():
"""Accented content should be found by unaccented query."""
user = factories.UserFactory()
dave = factories.BaseContactFactory(full_name="ééé", short_name="ààà")
client = APIClient()
client.force_login(user)
# Unaccented full name
response = client.get("/api/v1.0/contacts/?q=eee")
assert response.status_code == 200
contact_ids = [contact["id"] for contact in response.json()]
assert contact_ids == [str(dave.id)]
# Unaccented short name
response = client.get("/api/v1.0/contacts/?q=aaa")
assert response.status_code == 200
contact_ids = [contact["id"] for contact in response.json()]
assert contact_ids == [str(dave.id)]
def test_api_contacts_list_authenticated_accented_query():
"""Accented query should find unaccented content."""
user = factories.UserFactory()
dave = factories.BaseContactFactory(full_name="eee", short_name="aaa")
client = APIClient()
client.force_login(user)
# Unaccented full name
response = client.get("/api/v1.0/contacts/?q=ééé")
assert response.status_code == 200
contact_ids = [contact["id"] for contact in response.json()]
assert contact_ids == [str(dave.id)]
# Unaccented short name
response = client.get("/api/v1.0/contacts/?q=ààà")
assert response.status_code == 200
contact_ids = [contact["id"] for contact in response.json()]
assert contact_ids == [str(dave.id)]
@@ -1,89 +0,0 @@
"""
Test contacts API endpoints in People's core app.
"""
import pytest
from rest_framework.test import APIClient
from core import factories
pytestmark = pytest.mark.django_db
def test_api_contacts_retrieve_anonymous():
"""Anonymous users should not be allowed to retrieve a user."""
client = APIClient()
contact = factories.ContactFactory()
response = client.get(f"/api/v1.0/contacts/{contact.id!s}/")
assert response.status_code == 401
assert response.json() == {
"detail": "Authentication credentials were not provided."
}
def test_api_contacts_retrieve_authenticated_owned(django_assert_num_queries):
"""
Authenticated users should be allowed to retrieve a contact they own.
"""
user = factories.UserFactory()
contact = factories.ContactFactory(owner=user)
client = APIClient()
client.force_login(user)
with django_assert_num_queries(2):
response = client.get(f"/api/v1.0/contacts/{contact.id!s}/")
assert response.status_code == 200
assert response.json() == {
"id": str(contact.id),
"abilities": {"delete": True, "get": True, "patch": True, "put": True},
"override": None,
"owner": str(contact.owner.id),
"data": contact.data,
"full_name": contact.full_name,
"notes": "",
"short_name": contact.short_name,
}
def test_api_contacts_retrieve_authenticated_public():
"""
Authenticated users should be able to retrieve public contacts.
"""
user = factories.UserFactory()
contact = factories.BaseContactFactory()
client = APIClient()
client.force_login(user)
response = client.get(f"/api/v1.0/contacts/{contact.id!s}/")
assert response.status_code == 200
assert response.json() == {
"id": str(contact.id),
"abilities": {"delete": False, "get": True, "patch": False, "put": False},
"override": None,
"owner": None,
"data": contact.data,
"full_name": contact.full_name,
"notes": "",
"short_name": contact.short_name,
}
def test_api_contacts_retrieve_authenticated_other():
"""
Authenticated users should not be allowed to retrieve another user's contacts.
"""
user = factories.UserFactory()
contact = factories.ContactFactory()
client = APIClient()
client.force_login(user)
response = client.get(f"/api/v1.0/contacts/{contact.id!s}/")
assert response.status_code == 403
assert response.json() == {
"detail": "You do not have permission to perform this action."
}

Some files were not shown because too many files have changed in this diff Show More