5641674846

Committed 24 Jul 2023 06:42AM UTC coverage: 88.938% (-0.2%) from 89.184%

Build # 5641674846

Build Type

Pull #833

github

Committed by

web-flow

Commit Message

Merge 054f347f4 into 8c287ecf7

Pull Request Pull Request #833: Shared-cache

Run Details

1301 of 1301 new or added lines in 15 files covered. (100.0%)

105810 of 118970 relevant lines covered (88.94%)

60854.04 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

96.06

/services/src/pro/users/postgres_userdb.rs

use crate::contexts::SessionId;
use crate::error::Result;
use crate::pro::contexts::ProPostgresDb;
use crate::pro::permissions::{Role, RoleId};
use crate::pro::users::oidc::ExternalUserClaims;
use crate::pro::users::{
    User, UserCredentials, UserDb, UserId, UserInfo, UserRegistration, UserSession,
};
use crate::projects::{ProjectId, STRectangle};
use crate::util::Identifier;
use crate::{error, pro::contexts::ProPostgresContext};
use async_trait::async_trait;

use bb8_postgres::{
    tokio_postgres::tls::MakeTlsConnect, tokio_postgres::tls::TlsConnect, tokio_postgres::Socket,
};
use geoengine_datatypes::primitives::Duration;
use pwhash::bcrypt;
use snafu::ensure;
use uuid::Uuid;

use super::userdb::{RoleDb, UserAuth};

#[async_trait]
impl<Tls> UserAuth for ProPostgresContext<Tls>
where
    Tls: MakeTlsConnect<Socket> + Clone + Send + Sync + 'static,
    <Tls as MakeTlsConnect<Socket>>::Stream: Send + Sync,
    <Tls as MakeTlsConnect<Socket>>::TlsConnect: Send,
    <<Tls as MakeTlsConnect<Socket>>::TlsConnect as TlsConnect<Socket>>::Future: Send,
{
    // TODO: clean up expired sessions?

    async fn register_user(&self, user: UserRegistration) -> Result<UserId> {
        let mut conn = self.pool.get().await?;

        let tx = conn.build_transaction().start().await?;

        let user = User::from(user);

        let stmt = tx
            .prepare("INSERT INTO roles (id, name) VALUES ($1, $2);")
            .await?;
        tx.execute(&stmt, &[&user.id, &user.email]).await?;

        let stmt = tx
            .prepare(
                "INSERT INTO users (id, email, password_hash, real_name, quota_available, active) VALUES ($1, $2, $3, $4, $5, $6);",
            )
            .await?;

        let quota_available =
            crate::util::config::get_config_element::<crate::pro::util::config::Quota>()?
                .default_available_quota;

        tx.execute(
            &stmt,
            &[
                &user.id,
                &user.email,
                &user.password_hash,
                &user.real_name,
                &quota_available,
                &user.active,
            ],
        )
        .await?;

        let stmt = tx
            .prepare("INSERT INTO user_roles (user_id, role_id) VALUES ($1, $2);")
            .await?;
        tx.execute(&stmt, &[&user.id, &user.id]).await?;

        let stmt = tx
            .prepare("INSERT INTO user_roles (user_id, role_id) VALUES ($1, $2);")
            .await?;
        tx.execute(&stmt, &[&user.id, &Role::registered_user_role_id()])
            .await?;

        tx.commit().await?;

        Ok(user.id)
    }

    async fn create_anonymous_session(&self) -> Result<UserSession> {
        let mut conn = self.pool.get().await?;

        let tx = conn.build_transaction().start().await?;

        let user_id = UserId::new();

        let stmt = tx
            .prepare("INSERT INTO roles (id, name) VALUES ($1, $2);")
            .await?;
        tx.execute(&stmt, &[&user_id, &format!("anonymous_user_{user_id}")])
            .await?;

        let quota_available =
            crate::util::config::get_config_element::<crate::pro::util::config::Quota>()?
                .default_available_quota;

        let stmt = tx
            .prepare("INSERT INTO users (id, quota_available, active) VALUES ($1, $2, TRUE);")
            .await?;

        tx.execute(&stmt, &[&user_id, &quota_available]).await?;

        let stmt = tx
            .prepare("INSERT INTO user_roles (user_id, role_id) VALUES ($1, $2);")
            .await?;
        tx.execute(&stmt, &[&user_id, &user_id]).await?;

        let stmt = tx
            .prepare("INSERT INTO user_roles (user_id, role_id) VALUES ($1, $2);")
            .await?;
        tx.execute(&stmt, &[&user_id, &Role::anonymous_role_id()])
            .await?;

        let session_id = SessionId::new();
        let stmt = tx
            .prepare(
                "
                INSERT INTO sessions (id)
                VALUES ($1);",
            )
            .await?;

        tx.execute(&stmt, &[&session_id]).await?;

        let stmt = tx
            .prepare(
                "
            INSERT INTO 
                user_sessions (user_id, session_id, created, valid_until) 
            VALUES 
                ($1, $2, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP + make_interval(secs:=$3))
            RETURNING 
                created, valid_until;",
            )
            .await?;

        // TODO: load from config
        let session_duration = chrono::Duration::days(30);
        let row = tx
            .query_one(
                &stmt,
                &[
                    &user_id,
                    &session_id,
                    &(session_duration.num_seconds() as f64),
                ],
            )
            .await?;

        tx.commit().await?;

        Ok(UserSession {
            id: session_id,
            user: UserInfo {
                id: user_id,
                email: None,
                real_name: None,
            },
            created: row.get(0),
            valid_until: row.get(1),
            project: None,
            view: None,
            roles: vec![user_id.into(), Role::anonymous_role_id()],
        })
    }

    async fn login(&self, user_credentials: UserCredentials) -> Result<UserSession> {
        let mut conn = self.pool.get().await?;

        let tx = conn.build_transaction().start().await?;

        let stmt = tx
            .prepare("SELECT id, password_hash, email, real_name FROM users WHERE email = $1;")
            .await?;

        let row = tx
            .query_one(&stmt, &[&user_credentials.email])
            .await
            .map_err(|_error| error::Error::LoginFailed)?;

        let user_id = UserId(row.get(0));
        let password_hash = row.get(1);
        let email = row.get(2);
        let real_name = row.get(3);

        if bcrypt::verify(user_credentials.password, password_hash) {
            let session_id = SessionId::new();
            let stmt = tx
                .prepare(
                    "
                INSERT INTO sessions (id)
                VALUES ($1);",
                )
                .await?;

            tx.execute(&stmt, &[&session_id]).await?;

            // TODO: load from config
            let session_duration = chrono::Duration::days(30);
            let stmt = tx
                .prepare(
                    "
                INSERT INTO 
                    user_sessions (user_id, session_id, created, valid_until) 
                VALUES 
                    ($1, $2, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP + make_interval(secs:=$3))
                RETURNING 
                    created, valid_until;",
                )
                .await?;
            let row = tx
                .query_one(
                    &stmt,
                    &[
                        &user_id,
                        &session_id,
                        &(session_duration.num_seconds() as f64),
                    ],
                )
                .await?;

            let stmt = tx
                .prepare("SELECT role_id FROM user_roles WHERE user_id = $1;")
                .await?;

            let rows = tx
                .query(&stmt, &[&user_id])
                .await
                .map_err(|_error| error::Error::LoginFailed)?;

            tx.commit().await?;

            let roles = rows.into_iter().map(|row| row.get(0)).collect();

            Ok(UserSession {
                id: session_id,
                user: UserInfo {
                    id: user_id,
                    email,
                    real_name,
                },
                created: row.get(0),
                valid_until: row.get(1),
                project: None,
                view: None,
                roles,
            })
        } else {
            Err(error::Error::LoginFailed)
        }
    }

    #[allow(clippy::too_many_lines)]
    async fn login_external(
        &self,
        user: ExternalUserClaims,
        duration: Duration,
    ) -> Result<UserSession> {
        let mut conn = self.pool.get().await?;
        let tx = conn.build_transaction().start().await?;

        let stmt = tx
            .prepare("SELECT id, external_id, email, real_name FROM external_users WHERE external_id = $1;")
            .await?;

        let row = tx
            .query_opt(&stmt, &[&user.external_id.to_string()])
            .await
            .map_err(|_error| error::Error::LoginFailed)?;

        let user_id = match row {
            Some(row) => UserId(row.get(0)),
            None => {
                let user_id = UserId::new();

                let stmt = tx
                    .prepare("INSERT INTO roles (id, name) VALUES ($1, $2);")
                    .await?;
                tx.execute(&stmt, &[&user_id, &user.email]).await?;

                let quota_available =
                    crate::util::config::get_config_element::<crate::pro::util::config::Quota>()?
                        .default_available_quota;

                //TODO: Inconsistent to hashmap implementation, where an external user is not part of the user database.
                //TODO: A user might be able to login without external login using this (internal) id. Would be a problem with anonymous users as well.
                let stmt = tx
                    .prepare(
                        "INSERT INTO users (id, quota_available, active) VALUES ($1, $2, TRUE);",
                    )
                    .await?;
                tx.execute(&stmt, &[&user_id, &quota_available]).await?;

                let stmt = tx
                    .prepare(
                        "INSERT INTO external_users (id, external_id, email, real_name, active) VALUES ($1, $2, $3, $4, $5);",
                    )
                    .await?;

                tx.execute(
                    &stmt,
                    &[
                        &user_id,
                        &user.external_id.to_string(),
                        &user.email,
                        &user.real_name,
                        &true,
                    ],
                )
                .await?;

                let stmt = tx
                    .prepare("INSERT INTO user_roles (user_id, role_id) VALUES ($1, $2);")
                    .await?;
                tx.execute(&stmt, &[&user_id, &user_id]).await?;

                let stmt = tx
                    .prepare("INSERT INTO user_roles (user_id, role_id) VALUES ($1, $2);")
                    .await?;
                tx.execute(&stmt, &[&user_id, &Role::registered_user_role_id()])
                    .await?;

                user_id
            }
        };

        let session_id = SessionId::new();
        let stmt = tx
            .prepare(
                "
            INSERT INTO sessions (id)
            VALUES ($1);",
            )
            .await?; //TODO: Check documentation if inconsistent to hashmap implementation - would happen if CURRENT_TIMESTAMP is called twice in postgres for a single query. Worked in tests.

        tx.execute(&stmt, &[&session_id]).await?;

        let stmt = tx
            .prepare(
                "
            INSERT INTO 
                user_sessions (user_id, session_id, created, valid_until) 
            VALUES 
                ($1, $2, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP + make_interval(secs:=$3))
            RETURNING 
                created, valid_until;",
            )
            .await?;
        let row = tx
            .query_one(
                &stmt,
                &[&user_id, &session_id, &(duration.num_seconds() as f64)],
            )
            .await?;

        let stmt = tx
            .prepare("SELECT role_id FROM user_roles WHERE user_id = $1;")
            .await?;

        let rows = tx
            .query(&stmt, &[&user_id])
            .await
            .map_err(|_error| error::Error::LoginFailed)?;

        let roles = rows.into_iter().map(|row| row.get(0)).collect();

        tx.commit().await?;

        Ok(UserSession {
            id: session_id,
            user: UserInfo {
                id: user_id,
                email: Some(user.email.clone()),
                real_name: Some(user.real_name.clone()),
            },
            created: row.get(0),
            valid_until: row.get(1),
            project: None,
            view: None,
            roles,
        })
    }

    async fn user_session_by_id(&self, session: SessionId) -> Result<UserSession> {
        let mut conn = self.pool.get().await?;

        let tx = conn.build_transaction().start().await?;

        let stmt = tx
            .prepare(
                "
            SELECT 
                u.id,   
                u.email,
                u.real_name,             
                us.created, 
                us.valid_until, 
                s.project_id,
                s.view
            FROM 
                sessions s JOIN user_sessions us ON (s.id = us.session_id) 
                    JOIN users u ON (us.user_id = u.id)
            WHERE s.id = $1 AND CURRENT_TIMESTAMP < us.valid_until;",
            )
            .await?;

        let row = tx
            .query_one(&stmt, &[&session])
            .await
            .map_err(|_error| error::Error::InvalidSession)?;

        let mut session = UserSession {
            id: session,
            user: UserInfo {
                id: row.get(0),
                email: row.get(1),
                real_name: row.get(2),
            },
            created: row.get(3),
            valid_until: row.get(4),
            project: row.get::<usize, Option<Uuid>>(5).map(ProjectId),
            view: row.get(6),
            roles: vec![],
        };

        let stmt = tx
            .prepare(
                "
            SELECT role_id FROM user_roles WHERE user_id = $1;
            ",
            )
            .await?;

        let rows = tx.query(&stmt, &[&session.user.id]).await?;

        session.roles = rows.into_iter().map(|row| row.get(0)).collect();

        Ok(session)
    }
}

#[async_trait]
impl<Tls> UserDb for ProPostgresDb<Tls>
where
    Tls: MakeTlsConnect<Socket> + Clone + Send + Sync + 'static,
    <Tls as MakeTlsConnect<Socket>>::Stream: Send + Sync,
    <Tls as MakeTlsConnect<Socket>>::TlsConnect: Send,
    <<Tls as MakeTlsConnect<Socket>>::TlsConnect as TlsConnect<Socket>>::Future: Send,
{
    // TODO: clean up expired sessions?

    async fn logout(&self) -> Result<()> {
        let conn = self.conn_pool.get().await?;
        let stmt = conn
            .prepare("DELETE FROM sessions WHERE id = $1;") // TODO: only invalidate session?
            .await?;

        conn.execute(&stmt, &[&self.session.id])
            .await
            .map_err(|_error| error::Error::LogoutFailed)?;
        Ok(())
    }

    async fn set_session_project(&self, project: ProjectId) -> Result<()> {
        // TODO: check permission

        let conn = self.conn_pool.get().await?;
        let stmt = conn
            .prepare("UPDATE sessions SET project_id = $1 WHERE id = $2;")
            .await?;

        conn.execute(&stmt, &[&project, &self.session.id]).await?;

        Ok(())
    }

    async fn set_session_view(&self, view: STRectangle) -> Result<()> {
        let conn = self.conn_pool.get().await?;
        let stmt = conn
            .prepare("UPDATE sessions SET view = $1 WHERE id = $2;")
            .await?;

        conn.execute(&stmt, &[&view, &self.session.id]).await?;

        Ok(())
    }

    async fn increment_quota_used(&self, user: &UserId, quota_used: u64) -> Result<()> {
        ensure!(self.session.is_admin(), error::PermissionDenied);

        let conn = self.conn_pool.get().await?;
        let stmt = conn
            .prepare(
                "
            UPDATE users SET 
                quota_available = quota_available - $1, 
                quota_used = quota_used + $1
            WHERE id = $2;",
            )
            .await?;

        conn.execute(&stmt, &[&(quota_used as i64), &user]).await?;

        Ok(())
    }

    async fn quota_used(&self) -> Result<u64> {
        let conn = self.conn_pool.get().await?;
        let stmt = conn
            .prepare("SELECT quota_used FROM users WHERE id = $1;")
            .await?;

        let row = conn
            .query_one(&stmt, &[&self.session.user.id])
            .await
            .map_err(|_error| error::Error::InvalidSession)?;

        Ok(row.get::<usize, i64>(0) as u64)
    }

    async fn quota_used_by_user(&self, user: &UserId) -> Result<u64> {
        ensure!(
            self.session.user.id == *user || self.session.is_admin(),
            error::PermissionDenied
        );

        let conn = self.conn_pool.get().await?;
        let stmt = conn
            .prepare("SELECT quota_used FROM users WHERE id = $1;")
            .await?;

        let row = conn
            .query_one(&stmt, &[&user])
            .await
            .map_err(|_error| error::Error::InvalidSession)?;

        Ok(row.get::<usize, i64>(0) as u64)
    }

    async fn quota_available(&self) -> Result<i64> {
        let conn = self.conn_pool.get().await?;
        let stmt = conn
            .prepare("SELECT quota_available FROM users WHERE id = $1;")
            .await?;

        let row = conn
            .query_one(&stmt, &[&self.session.user.id])
            .await
            .map_err(|_error| error::Error::InvalidSession)?;

        Ok(row.get::<usize, i64>(0))
    }

    async fn quota_available_by_user(&self, user: &UserId) -> Result<i64> {
        ensure!(
            self.session.user.id == *user || self.session.is_admin(),
            error::PermissionDenied
        );

        let conn = self.conn_pool.get().await?;
        let stmt = conn
            .prepare("SELECT quota_available FROM users WHERE id = $1;")
            .await?;

        let row = conn
            .query_one(&stmt, &[&user])
            .await
            .map_err(|_error| error::Error::InvalidSession)?;

        Ok(row.get::<usize, i64>(0))
    }

    async fn update_quota_available_by_user(
        &self,
        user: &UserId,
        new_available_quota: i64,
    ) -> Result<()> {
        ensure!(self.session.is_admin(), error::PermissionDenied);

        let conn = self.conn_pool.get().await?;
        let stmt = conn
            .prepare(
                "
            UPDATE users SET 
                quota_available = $1
            WHERE id = $2;",
            )
            .await?;

        conn.execute(&stmt, &[&(new_available_quota), &user])
            .await?;

        Ok(())
    }
}

#[async_trait]
impl<Tls> RoleDb for ProPostgresDb<Tls>
where
    Tls: MakeTlsConnect<Socket> + Clone + Send + Sync + 'static,
    <Tls as MakeTlsConnect<Socket>>::Stream: Send + Sync,
    <Tls as MakeTlsConnect<Socket>>::TlsConnect: Send,
    <<Tls as MakeTlsConnect<Socket>>::TlsConnect as TlsConnect<Socket>>::Future: Send,
{
    async fn add_role(&self, role_name: &str) -> Result<RoleId> {
        ensure!(self.session.is_admin(), error::PermissionDenied);

        let conn = self.conn_pool.get().await?;

        let id = RoleId::new();

        let stmt = conn
            .prepare("INSERT INTO roles (id, name) VALUES ($1, $2);")
            .await?;

        // TODO: map postgres error code to error::Error::RoleAlreadyExists

        conn.execute(&stmt, &[&id, &role_name]).await?;

        Ok(id)
    }

    async fn remove_role(&self, role_id: &RoleId) -> Result<()> {
        ensure!(self.session.is_admin(), error::PermissionDenied);

        let conn = self.conn_pool.get().await?;

        let stmt = conn.prepare("DELETE FROM roles WHERE id = $1;").await?;

        let deleted = conn.execute(&stmt, &[&role_id]).await?;

        ensure!(deleted > 0, error::RoleDoesNotExist);

        Ok(())
    }

    async fn assign_role(&self, role_id: &RoleId, user_id: &UserId) -> Result<()> {
        ensure!(self.session.is_admin(), error::PermissionDenied);

        let conn = self.conn_pool.get().await?;

        let stmt = conn
            .prepare("INSERT INTO user_roles (user_id, role_id) VALUES ($1, $2);")
            .await?;

        // TODO: map postgres error code to error::Error::RoleAlreadyAssigned, RoleDoesNotExist

        conn.execute(&stmt, &[&user_id, &role_id]).await?;

        Ok(())
    }

    async fn revoke_role(&self, role_id: &RoleId, user_id: &UserId) -> Result<()> {
        ensure!(self.session.is_admin(), error::PermissionDenied);

        let conn = self.conn_pool.get().await?;

        let stmt = conn
            .prepare("DELETE FROM user_roles WHERE user_id= $1 AND role_id = $2;")
            .await?;

        let deleted = conn.execute(&stmt, &[&user_id, &role_id]).await?;

        ensure!(deleted > 0, error::RoleNotAssigned);

        Ok(())
    }
}

1	use crate::contexts::SessionId;
2	use crate::error::Result;
3	use crate::pro::contexts::ProPostgresDb;
4	use crate::pro::permissions::{Role, RoleId};
5	use crate::pro::users::oidc::ExternalUserClaims;
6	use crate::pro::users::{
7	User, UserCredentials, UserDb, UserId, UserInfo, UserRegistration, UserSession,
8	};
9	use crate::projects::{ProjectId, STRectangle};
10	use crate::util::Identifier;
11	use crate::{error, pro::contexts::ProPostgresContext};
12	use async_trait::async_trait;
13
14	use bb8_postgres::{
15	tokio_postgres::tls::MakeTlsConnect, tokio_postgres::tls::TlsConnect, tokio_postgres::Socket,
16	};
17	use geoengine_datatypes::primitives::Duration;
18	use pwhash::bcrypt;
19	use snafu::ensure;
20	use uuid::Uuid;
21
22	use super::userdb::{RoleDb, UserAuth};
23
24	#[async_trait]
25	impl<Tls> UserAuth for ProPostgresContext<Tls>
26	where
27	Tls: MakeTlsConnect<Socket> + Clone + Send + Sync + 'static,
28	<Tls as MakeTlsConnect<Socket>>::Stream: Send + Sync,
29	<Tls as MakeTlsConnect<Socket>>::TlsConnect: Send,
30	<<Tls as MakeTlsConnect<Socket>>::TlsConnect as TlsConnect<Socket>>::Future: Send,
31	{
32	// TODO: clean up expired sessions?
33
34	async fn register_user(&self, user: UserRegistration) -> Result<UserId> {	7✔
35	let mut conn = self.pool.get().await?;	7✔
36
37	let tx = conn.build_transaction().start().await?;	7✔
38
39	let user = User::from(user);	7✔
40
41	let stmt = tx	7✔
42	.prepare("INSERT INTO roles (id, name) VALUES ($1, $2);")	7✔
43	.await?;	5✔
44	tx.execute(&stmt, &[&user.id, &user.email]).await?;	7✔
45
46	let stmt = tx	7✔
47	.prepare(	7✔
48	"INSERT INTO users (id, email, password_hash, real_name, quota_available, active) VALUES ($1, $2, $3, $4, $5, $6);",	7✔
49	)	7✔
50	.await?;	4✔
51
52	let quota_available =	7✔
53	crate::util::config::get_config_element::<crate::pro::util::config::Quota>()?	7✔
54	.default_available_quota;
55
56	tx.execute(	7✔
57	&stmt,	7✔
58	&[	7✔
59	&user.id,	7✔
60	&user.email,	7✔
61	&user.password_hash,	7✔
62	&user.real_name,	7✔
63	&quota_available,	7✔
64	&user.active,	7✔
65	],	7✔
66	)	7✔
67	.await?;	3✔
68
69	let stmt = tx	7✔
70	.prepare("INSERT INTO user_roles (user_id, role_id) VALUES ($1, $2);")	7✔
71	.await?;	4✔
72	tx.execute(&stmt, &[&user.id, &user.id]).await?;	7✔
73
74	let stmt = tx	7✔
75	.prepare("INSERT INTO user_roles (user_id, role_id) VALUES ($1, $2);")	7✔
76	.await?;	4✔
77	tx.execute(&stmt, &[&user.id, &Role::registered_user_role_id()])	7✔
78	.await?;	4✔
79
80	tx.commit().await?;	7✔
81
82	Ok(user.id)	7✔
83	}	14✔
84
85	async fn create_anonymous_session(&self) -> Result<UserSession> {	21✔
86	let mut conn = self.pool.get().await?;	21✔
87
88	let tx = conn.build_transaction().start().await?;	21✔
89
90	let user_id = UserId::new();	21✔
91
92	let stmt = tx	21✔
93	.prepare("INSERT INTO roles (id, name) VALUES ($1, $2);")	21✔
94	.await?;	14✔
95	tx.execute(&stmt, &[&user_id, &format!("anonymous_user_{user_id}")])	21✔
96	.await?;	14✔
97
98	let quota_available =	21✔
99	crate::util::config::get_config_element::<crate::pro::util::config::Quota>()?	21✔
100	.default_available_quota;
101
102	let stmt = tx	21✔
103	.prepare("INSERT INTO users (id, quota_available, active) VALUES ($1, $2, TRUE);")	21✔
104	.await?;	14✔
105
106	tx.execute(&stmt, &[&user_id, &quota_available]).await?;	21✔
107
108	let stmt = tx	21✔
109	.prepare("INSERT INTO user_roles (user_id, role_id) VALUES ($1, $2);")	21✔
110	.await?;	16✔
111	tx.execute(&stmt, &[&user_id, &user_id]).await?;	21✔
112
113	let stmt = tx	21✔
114	.prepare("INSERT INTO user_roles (user_id, role_id) VALUES ($1, $2);")	21✔
115	.await?;	16✔
116	tx.execute(&stmt, &[&user_id, &Role::anonymous_role_id()])	21✔
117	.await?;	16✔
118
119	let session_id = SessionId::new();	21✔
120	let stmt = tx	21✔
121	.prepare(	21✔
122	"	21✔
123	INSERT INTO sessions (id)	21✔
124	VALUES ($1);",	21✔
125	)	21✔
126	.await?;	17✔
127
128	tx.execute(&stmt, &[&session_id]).await?;	21✔
129
130	let stmt = tx	21✔
131	.prepare(	21✔
132	"	21✔
133	INSERT INTO	21✔
134	user_sessions (user_id, session_id, created, valid_until)	21✔
135	VALUES	21✔
136	($1, $2, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP + make_interval(secs:=$3))	21✔
137	RETURNING	21✔
138	created, valid_until;",	21✔
139	)	21✔
140	.await?;	16✔
141
142	// TODO: load from config
143	let session_duration = chrono::Duration::days(30);	21✔
144	let row = tx	21✔
145	.query_one(	21✔
146	&stmt,	21✔
147	&[	21✔
148	&user_id,	21✔
149	&session_id,	21✔
150	&(session_duration.num_seconds() as f64),	21✔
151	],	21✔
152	)	21✔
153	.await?;	17✔
154
155	tx.commit().await?;	21✔
156
157	Ok(UserSession {	21✔
158	id: session_id,	21✔
159	user: UserInfo {	21✔
160	id: user_id,	21✔
161	email: None,	21✔
162	real_name: None,	21✔
163	},	21✔
164	created: row.get(0),	21✔
165	valid_until: row.get(1),	21✔
166	project: None,	21✔
167	view: None,	21✔
168	roles: vec![user_id.into(), Role::anonymous_role_id()],	21✔
169	})	21✔
170	}	42✔
171
172	async fn login(&self, user_credentials: UserCredentials) -> Result<UserSession> {	22✔
173	let mut conn = self.pool.get().await?;	22✔
174
175	let tx = conn.build_transaction().start().await?;	22✔
176
177	let stmt = tx	22✔
178	.prepare("SELECT id, password_hash, email, real_name FROM users WHERE email = $1;")	22✔
179	.await?;	17✔
180
181	let row = tx	22✔
182	.query_one(&stmt, &[&user_credentials.email])	22✔
183	.await	15✔
184	.map_err(\|_error\| error::Error::LoginFailed)?;	22✔
185
186	let user_id = UserId(row.get(0));	22✔
187	let password_hash = row.get(1);	22✔
188	let email = row.get(2);	22✔
189	let real_name = row.get(3);	22✔
190		22✔
191	if bcrypt::verify(user_credentials.password, password_hash) {	22✔
192	let session_id = SessionId::new();	22✔
193	let stmt = tx	22✔
194	.prepare(	22✔
195	"	22✔
196	INSERT INTO sessions (id)	22✔
197	VALUES ($1);",	22✔
198	)	22✔
199	.await?;	14✔
200
201	tx.execute(&stmt, &[&session_id]).await?;	22✔
202
203	// TODO: load from config
204	let session_duration = chrono::Duration::days(30);	22✔
205	let stmt = tx	22✔
206	.prepare(	22✔
207	"	22✔
208	INSERT INTO	22✔
209	user_sessions (user_id, session_id, created, valid_until)	22✔
210	VALUES	22✔
211	($1, $2, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP + make_interval(secs:=$3))	22✔
212	RETURNING	22✔
213	created, valid_until;",	22✔
214	)	22✔
215	.await?;	14✔
216	let row = tx	22✔
217	.query_one(	22✔
218	&stmt,	22✔
219	&[	22✔
220	&user_id,	22✔
221	&session_id,	22✔
222	&(session_duration.num_seconds() as f64),	22✔
223	],	22✔
224	)	22✔
225	.await?;	13✔
226
227	let stmt = tx	22✔
228	.prepare("SELECT role_id FROM user_roles WHERE user_id = $1;")	22✔
229	.await?;	14✔
230
231	let rows = tx	22✔
232	.query(&stmt, &[&user_id])	22✔
233	.await	14✔
234	.map_err(\|_error\| error::Error::LoginFailed)?;	22✔
235
236	tx.commit().await?;	22✔
237
238	let roles = rows.into_iter().map(\|row\| row.get(0)).collect();	35✔
239		22✔
240	Ok(UserSession {	22✔
241	id: session_id,	22✔
242	user: UserInfo {	22✔
243	id: user_id,	22✔
244	email,	22✔
245	real_name,	22✔
246	},	22✔
247	created: row.get(0),	22✔
248	valid_until: row.get(1),	22✔
249	project: None,	22✔
250	view: None,	22✔
251	roles,	22✔
252	})	22✔
253	} else {
254	Err(error::Error::LoginFailed)	×
255	}
256	}	44✔
257
258	#[allow(clippy::too_many_lines)]
259	async fn login_external(	3✔
260	&self,	3✔
261	user: ExternalUserClaims,	3✔
262	duration: Duration,	3✔
263	) -> Result<UserSession> {	3✔
264	let mut conn = self.pool.get().await?;	3✔
265	let tx = conn.build_transaction().start().await?;	3✔
266
267	let stmt = tx	3✔
268	.prepare("SELECT id, external_id, email, real_name FROM external_users WHERE external_id = $1;")	3✔
269	.await?;	3✔
270
271	let row = tx	3✔
272	.query_opt(&stmt, &[&user.external_id.to_string()])	3✔
273	.await	3✔
274	.map_err(\|_error\| error::Error::LoginFailed)?;	3✔
275
276	let user_id = match row {	3✔
277	Some(row) => UserId(row.get(0)),	2✔
278	None => {
279	let user_id = UserId::new();	1✔
280
281	let stmt = tx	1✔
282	.prepare("INSERT INTO roles (id, name) VALUES ($1, $2);")	1✔
283	.await?;	1✔
284	tx.execute(&stmt, &[&user_id, &user.email]).await?;	1✔
285
286	let quota_available =	1✔
287	crate::util::config::get_config_element::<crate::pro::util::config::Quota>()?	1✔
288	.default_available_quota;
289
290	//TODO: Inconsistent to hashmap implementation, where an external user is not part of the user database.
291	//TODO: A user might be able to login without external login using this (internal) id. Would be a problem with anonymous users as well.
292	let stmt = tx	1✔
293	.prepare(	1✔
294	"INSERT INTO users (id, quota_available, active) VALUES ($1, $2, TRUE);",	1✔
295	)	1✔
296	.await?;	1✔
297	tx.execute(&stmt, &[&user_id, &quota_available]).await?;	1✔
298
299	let stmt = tx	1✔
300	.prepare(	1✔
301	"INSERT INTO external_users (id, external_id, email, real_name, active) VALUES ($1, $2, $3, $4, $5);",	1✔
302	)	1✔
303	.await?;	1✔
304
305	tx.execute(	1✔
306	&stmt,	1✔
307	&[	1✔
308	&user_id,	1✔
309	&user.external_id.to_string(),	1✔
310	&user.email,	1✔
311	&user.real_name,	1✔
312	&true,	1✔
313	],	1✔
314	)	1✔
315	.await?;	1✔
316
317	let stmt = tx	1✔
318	.prepare("INSERT INTO user_roles (user_id, role_id) VALUES ($1, $2);")	1✔
319	.await?;	1✔
320	tx.execute(&stmt, &[&user_id, &user_id]).await?;	1✔
321
322	let stmt = tx	1✔
323	.prepare("INSERT INTO user_roles (user_id, role_id) VALUES ($1, $2);")	1✔
324	.await?;	1✔
325	tx.execute(&stmt, &[&user_id, &Role::registered_user_role_id()])	1✔
326	.await?;	1✔
327
328	user_id	1✔
329	}
330	};
331
332	let session_id = SessionId::new();	3✔
333	let stmt = tx	3✔
334	.prepare(	3✔
335	"	3✔
336	INSERT INTO sessions (id)	3✔
337	VALUES ($1);",	3✔
338	)	3✔
339	.await?; //TODO: Check documentation if inconsistent to hashmap implementation - would happen if CURRENT_TIMESTAMP is called twice in postgres for a single query. Worked in tests.	3✔
340
341	tx.execute(&stmt, &[&session_id]).await?;	3✔
342
343	let stmt = tx	3✔
344	.prepare(	3✔
345	"	3✔
346	INSERT INTO	3✔
347	user_sessions (user_id, session_id, created, valid_until)	3✔
348	VALUES	3✔
349	($1, $2, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP + make_interval(secs:=$3))	3✔
350	RETURNING	3✔
351	created, valid_until;",	3✔
352	)	3✔
353	.await?;	3✔
354	let row = tx	3✔
355	.query_one(	3✔
356	&stmt,	3✔
357	&[&user_id, &session_id, &(duration.num_seconds() as f64)],	3✔
358	)	3✔
359	.await?;	3✔
360
361	let stmt = tx	3✔
362	.prepare("SELECT role_id FROM user_roles WHERE user_id = $1;")	3✔
363	.await?;	3✔
364
365	let rows = tx	3✔
366	.query(&stmt, &[&user_id])	3✔
367	.await	3✔
368	.map_err(\|_error\| error::Error::LoginFailed)?;	3✔
369
370	let roles = rows.into_iter().map(\|row\| row.get(0)).collect();	6✔
371		3✔
372	tx.commit().await?;	3✔
373
374	Ok(UserSession {	3✔
375	id: session_id,	3✔
376	user: UserInfo {	3✔
377	id: user_id,	3✔
378	email: Some(user.email.clone()),	3✔
379	real_name: Some(user.real_name.clone()),	3✔
380	},	3✔
381	created: row.get(0),	3✔
382	valid_until: row.get(1),	3✔
383	project: None,	3✔
384	view: None,	3✔
385	roles,	3✔
386	})	3✔
387	}	6✔
388
389	async fn user_session_by_id(&self, session: SessionId) -> Result<UserSession> {	13✔
390	let mut conn = self.pool.get().await?;	13✔
391
392	let tx = conn.build_transaction().start().await?;	13✔
393
394	let stmt = tx	13✔
395	.prepare(	13✔
396	"	13✔
397	SELECT	13✔
398	u.id,	13✔
399	u.email,	13✔
400	u.real_name,	13✔
401	us.created,	13✔
402	us.valid_until,	13✔
403	s.project_id,	13✔
404	s.view	13✔
405	FROM	13✔
406	sessions s JOIN user_sessions us ON (s.id = us.session_id)	13✔
407	JOIN users u ON (us.user_id = u.id)	13✔
408	WHERE s.id = $1 AND CURRENT_TIMESTAMP < us.valid_until;",	13✔
409	)	13✔
410	.await?;	43✔
411
412	let row = tx	13✔
413	.query_one(&stmt, &[&session])	13✔
414	.await	13✔
415	.map_err(\|_error\| error::Error::InvalidSession)?;	13✔
416
417	let mut session = UserSession {	9✔
418	id: session,	9✔
419	user: UserInfo {	9✔
420	id: row.get(0),	9✔
421	email: row.get(1),	9✔
422	real_name: row.get(2),	9✔
423	},	9✔
424	created: row.get(3),	9✔
425	valid_until: row.get(4),	9✔
426	project: row.get::<usize, Option<Uuid>>(5).map(ProjectId),	9✔
427	view: row.get(6),	9✔
428	roles: vec![],	9✔
429	};	9✔
430
431	let stmt = tx	9✔
432	.prepare(	9✔
433	"	9✔
434	SELECT role_id FROM user_roles WHERE user_id = $1;	9✔
435	",	9✔
436	)	9✔
437	.await?;	9✔
438
439	let rows = tx.query(&stmt, &[&session.user.id]).await?;	9✔
440
441	session.roles = rows.into_iter().map(\|row\| row.get(0)).collect();	18✔
442		9✔
443	Ok(session)	9✔
444	}	26✔
445	}
446
447	#[async_trait]
448	impl<Tls> UserDb for ProPostgresDb<Tls>
449	where
450	Tls: MakeTlsConnect<Socket> + Clone + Send + Sync + 'static,
451	<Tls as MakeTlsConnect<Socket>>::Stream: Send + Sync,
452	<Tls as MakeTlsConnect<Socket>>::TlsConnect: Send,
453	<<Tls as MakeTlsConnect<Socket>>::TlsConnect as TlsConnect<Socket>>::Future: Send,
454	{
455	// TODO: clean up expired sessions?
456
457	async fn logout(&self) -> Result<()> {	4✔
458	let conn = self.conn_pool.get().await?;	4✔
459	let stmt = conn	4✔
460	.prepare("DELETE FROM sessions WHERE id = $1;") // TODO: only invalidate session?	4✔
461	.await?;	4✔
462
463	conn.execute(&stmt, &[&self.session.id])	4✔
464	.await	4✔
465	.map_err(\|_error\| error::Error::LogoutFailed)?;	4✔
466	Ok(())	4✔
467	}	8✔
468
469	async fn set_session_project(&self, project: ProjectId) -> Result<()> {	2✔
470	// TODO: check permission
471
472	let conn = self.conn_pool.get().await?;	2✔
473	let stmt = conn	2✔
474	.prepare("UPDATE sessions SET project_id = $1 WHERE id = $2;")	2✔
475	.await?;	2✔
476
477	conn.execute(&stmt, &[&project, &self.session.id]).await?;	2✔
478
479	Ok(())	2✔
480	}	4✔
481
482	async fn set_session_view(&self, view: STRectangle) -> Result<()> {	2✔
483	let conn = self.conn_pool.get().await?;	2✔
484	let stmt = conn	2✔
485	.prepare("UPDATE sessions SET view = $1 WHERE id = $2;")	2✔
486	.await?;	2✔
487
488	conn.execute(&stmt, &[&view, &self.session.id]).await?;	2✔
489
490	Ok(())	2✔
491	}	4✔
492
493	async fn increment_quota_used(&self, user: &UserId, quota_used: u64) -> Result<()> {	4✔
494	ensure!(self.session.is_admin(), error::PermissionDenied);	4✔
495
496	let conn = self.conn_pool.get().await?;	4✔
497	let stmt = conn	4✔
498	.prepare(	4✔
499	"	4✔
500	UPDATE users SET	4✔
501	quota_available = quota_available - $1,	4✔
502	quota_used = quota_used + $1	4✔
503	WHERE id = $2;",	4✔
504	)	4✔
505	.await?;	4✔
506
507	conn.execute(&stmt, &[&(quota_used as i64), &user]).await?;	4✔
508
509	Ok(())	4✔
510	}	8✔
511
512	async fn quota_used(&self) -> Result<u64> {	3✔
513	let conn = self.conn_pool.get().await?;	3✔
514	let stmt = conn	3✔
515	.prepare("SELECT quota_used FROM users WHERE id = $1;")	3✔
516	.await?;	3✔
517
518	let row = conn	3✔
519	.query_one(&stmt, &[&self.session.user.id])	3✔
520	.await	3✔
521	.map_err(\|_error\| error::Error::InvalidSession)?;	3✔
522
523	Ok(row.get::<usize, i64>(0) as u64)	3✔
524	}	6✔
525
526	async fn quota_used_by_user(&self, user: &UserId) -> Result<u64> {	×
527	ensure!(	×
528	self.session.user.id == *user \|\| self.session.is_admin(),	×
529	error::PermissionDenied	×
530	);
531
532	let conn = self.conn_pool.get().await?;	×
533	let stmt = conn	×
534	.prepare("SELECT quota_used FROM users WHERE id = $1;")	×
535	.await?;	×
536
537	let row = conn	×
538	.query_one(&stmt, &[&user])	×
539	.await	×
540	.map_err(\|_error\| error::Error::InvalidSession)?;	×
541
542	Ok(row.get::<usize, i64>(0) as u64)	×
543	}	×
544
545	async fn quota_available(&self) -> Result<i64> {	4✔
546	let conn = self.conn_pool.get().await?;	4✔
547	let stmt = conn	4✔
548	.prepare("SELECT quota_available FROM users WHERE id = $1;")	4✔
549	.await?;	1✔
550
551	let row = conn	4✔
552	.query_one(&stmt, &[&self.session.user.id])	4✔
553	.await	2✔
554	.map_err(\|_error\| error::Error::InvalidSession)?;	4✔
555
556	Ok(row.get::<usize, i64>(0))	4✔
557	}	8✔
558
559	async fn quota_available_by_user(&self, user: &UserId) -> Result<i64> {	2✔
560	ensure!(	2✔
561	self.session.user.id == *user \|\| self.session.is_admin(),	2✔
562	error::PermissionDenied	×
563	);
564
565	let conn = self.conn_pool.get().await?;	2✔
566	let stmt = conn	2✔
567	.prepare("SELECT quota_available FROM users WHERE id = $1;")	2✔
568	.await?;	×
569
570	let row = conn	2✔
571	.query_one(&stmt, &[&user])	2✔
572	.await	×
573	.map_err(\|_error\| error::Error::InvalidSession)?;	2✔
574
575	Ok(row.get::<usize, i64>(0))	2✔
576	}	4✔
577
578	async fn update_quota_available_by_user(	2✔
579	&self,	2✔
580	user: &UserId,	2✔
581	new_available_quota: i64,	2✔
582	) -> Result<()> {	2✔
583	ensure!(self.session.is_admin(), error::PermissionDenied);	2✔
584
585	let conn = self.conn_pool.get().await?;	2✔
586	let stmt = conn	2✔
587	.prepare(	2✔
588	"	2✔
589	UPDATE users SET	2✔
590	quota_available = $1	2✔
591	WHERE id = $2;",	2✔
592	)	2✔
593	.await?;	×
594
595	conn.execute(&stmt, &[&(new_available_quota), &user])	2✔
596	.await?;	2✔
597
598	Ok(())	2✔
599	}	4✔
600	}
601
602	#[async_trait]
603	impl<Tls> RoleDb for ProPostgresDb<Tls>
604	where
605	Tls: MakeTlsConnect<Socket> + Clone + Send + Sync + 'static,
606	<Tls as MakeTlsConnect<Socket>>::Stream: Send + Sync,
607	<Tls as MakeTlsConnect<Socket>>::TlsConnect: Send,
608	<<Tls as MakeTlsConnect<Socket>>::TlsConnect as TlsConnect<Socket>>::Future: Send,
609	{
610	async fn add_role(&self, role_name: &str) -> Result<RoleId> {	1✔
611	ensure!(self.session.is_admin(), error::PermissionDenied);	1✔
612
613	let conn = self.conn_pool.get().await?;	1✔
614
615	let id = RoleId::new();	1✔
616
617	let stmt = conn	1✔
618	.prepare("INSERT INTO roles (id, name) VALUES ($1, $2);")	1✔
619	.await?;	1✔
620
621	// TODO: map postgres error code to error::Error::RoleAlreadyExists
622
623	conn.execute(&stmt, &[&id, &role_name]).await?;	1✔
624
625	Ok(id)	1✔
626	}	2✔
627
628	async fn remove_role(&self, role_id: &RoleId) -> Result<()> {	1✔
629	ensure!(self.session.is_admin(), error::PermissionDenied);	1✔
630
631	let conn = self.conn_pool.get().await?;	1✔
632
633	let stmt = conn.prepare("DELETE FROM roles WHERE id = $1;").await?;	1✔
634
635	let deleted = conn.execute(&stmt, &[&role_id]).await?;	1✔
636
637	ensure!(deleted > 0, error::RoleDoesNotExist);	1✔
638
639	Ok(())	1✔
640	}	2✔
641
642	async fn assign_role(&self, role_id: &RoleId, user_id: &UserId) -> Result<()> {	2✔
643	ensure!(self.session.is_admin(), error::PermissionDenied);	2✔
644
645	let conn = self.conn_pool.get().await?;	2✔
646
647	let stmt = conn	2✔
648	.prepare("INSERT INTO user_roles (user_id, role_id) VALUES ($1, $2);")	2✔
649	.await?;	2✔
650
651	// TODO: map postgres error code to error::Error::RoleAlreadyAssigned, RoleDoesNotExist
652
653	conn.execute(&stmt, &[&user_id, &role_id]).await?;	2✔
654
655	Ok(())	2✔
656	}	4✔
657
658	async fn revoke_role(&self, role_id: &RoleId, user_id: &UserId) -> Result<()> {	1✔
659	ensure!(self.session.is_admin(), error::PermissionDenied);	1✔
660
661	let conn = self.conn_pool.get().await?;	1✔
662
663	let stmt = conn	1✔
664	.prepare("DELETE FROM user_roles WHERE user_id= $1 AND role_id = $2;")	1✔
665	.await?;	1✔
666
667	let deleted = conn.execute(&stmt, &[&user_id, &role_id]).await?;	1✔
668
669	ensure!(deleted > 0, error::RoleNotAssigned);	1✔
670
671	Ok(())	1✔
672	}	2✔
673	}

geo-engine / geoengine / 5641674846

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous