123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455 |
- // Copyright 2018 The Gitea Authors. All rights reserved.
- // SPDX-License-Identifier: MIT
-
- package access
-
- import (
- "context"
- "fmt"
- "slices"
-
- "code.gitea.io/gitea/models/db"
- "code.gitea.io/gitea/models/organization"
- perm_model "code.gitea.io/gitea/models/perm"
- repo_model "code.gitea.io/gitea/models/repo"
- "code.gitea.io/gitea/models/unit"
- user_model "code.gitea.io/gitea/models/user"
- "code.gitea.io/gitea/modules/log"
- "code.gitea.io/gitea/modules/util"
- )
-
- // Permission contains all the permissions related variables to a repository for a user
- type Permission struct {
- AccessMode perm_model.AccessMode
-
- units []*repo_model.RepoUnit
- unitsMode map[unit.Type]perm_model.AccessMode
- }
-
- // IsOwner returns true if current user is the owner of repository.
- func (p *Permission) IsOwner() bool {
- return p.AccessMode >= perm_model.AccessModeOwner
- }
-
- // IsAdmin returns true if current user has admin or higher access of repository.
- func (p *Permission) IsAdmin() bool {
- return p.AccessMode >= perm_model.AccessModeAdmin
- }
-
- // HasAccess returns true if the current user might have at least read access to any unit of this repository
- func (p *Permission) HasAccess() bool {
- return len(p.unitsMode) > 0 || p.AccessMode >= perm_model.AccessModeRead
- }
-
- // HasUnits returns true if the permission contains attached units
- func (p *Permission) HasUnits() bool {
- return len(p.units) > 0
- }
-
- // GetFirstUnitRepoID returns the repo ID of the first unit, it is a fragile design and should NOT be used anymore
- // deprecated
- func (p *Permission) GetFirstUnitRepoID() int64 {
- if len(p.units) > 0 {
- return p.units[0].RepoID
- }
- return 0
- }
-
- // UnitAccessMode returns current user access mode to the specify unit of the repository
- func (p *Permission) UnitAccessMode(unitType unit.Type) perm_model.AccessMode {
- if p.unitsMode != nil {
- // if the units map contains the access mode, use it, but admin/owner mode could override it
- if m, ok := p.unitsMode[unitType]; ok {
- return util.Iif(p.AccessMode >= perm_model.AccessModeAdmin, p.AccessMode, m)
- }
- }
- // if the units map does not contain the access mode, return the default access mode if the unit exists
- hasUnit := slices.ContainsFunc(p.units, func(u *repo_model.RepoUnit) bool { return u.Type == unitType })
- return util.Iif(hasUnit, p.AccessMode, perm_model.AccessModeNone)
- }
-
- func (p *Permission) SetUnitsWithDefaultAccessMode(units []*repo_model.RepoUnit, mode perm_model.AccessMode) {
- p.units = units
- p.unitsMode = make(map[unit.Type]perm_model.AccessMode)
- for _, u := range p.units {
- p.unitsMode[u.Type] = mode
- }
- }
-
- // CanAccess returns true if user has mode access to the unit of the repository
- func (p *Permission) CanAccess(mode perm_model.AccessMode, unitType unit.Type) bool {
- return p.UnitAccessMode(unitType) >= mode
- }
-
- // CanAccessAny returns true if user has mode access to any of the units of the repository
- func (p *Permission) CanAccessAny(mode perm_model.AccessMode, unitTypes ...unit.Type) bool {
- for _, u := range unitTypes {
- if p.CanAccess(mode, u) {
- return true
- }
- }
- return false
- }
-
- // CanRead returns true if user could read to this unit
- func (p *Permission) CanRead(unitType unit.Type) bool {
- return p.CanAccess(perm_model.AccessModeRead, unitType)
- }
-
- // CanReadAny returns true if user has read access to any of the units of the repository
- func (p *Permission) CanReadAny(unitTypes ...unit.Type) bool {
- return p.CanAccessAny(perm_model.AccessModeRead, unitTypes...)
- }
-
- // CanReadIssuesOrPulls returns true if isPull is true and user could read pull requests and
- // returns true if isPull is false and user could read to issues
- func (p *Permission) CanReadIssuesOrPulls(isPull bool) bool {
- if isPull {
- return p.CanRead(unit.TypePullRequests)
- }
- return p.CanRead(unit.TypeIssues)
- }
-
- // CanWrite returns true if user could write to this unit
- func (p *Permission) CanWrite(unitType unit.Type) bool {
- return p.CanAccess(perm_model.AccessModeWrite, unitType)
- }
-
- // CanWriteIssuesOrPulls returns true if isPull is true and user could write to pull requests and
- // returns true if isPull is false and user could write to issues
- func (p *Permission) CanWriteIssuesOrPulls(isPull bool) bool {
- if isPull {
- return p.CanWrite(unit.TypePullRequests)
- }
- return p.CanWrite(unit.TypeIssues)
- }
-
- func (p *Permission) ReadableUnitTypes() []unit.Type {
- types := make([]unit.Type, 0, len(p.units))
- for _, u := range p.units {
- if p.CanRead(u.Type) {
- types = append(types, u.Type)
- }
- }
- return types
- }
-
- func (p *Permission) LogString() string {
- format := "<Permission AccessMode=%s, %d Units, %d UnitsMode(s): [ "
- args := []any{p.AccessMode.ToString(), len(p.units), len(p.unitsMode)}
-
- for i, u := range p.units {
- config := ""
- if u.Config != nil {
- configBytes, err := u.Config.ToDB()
- config = string(configBytes)
- if err != nil {
- config = err.Error()
- }
- }
- format += "\nUnits[%d]: ID: %d RepoID: %d Type: %s Config: %s"
- args = append(args, i, u.ID, u.RepoID, u.Type.LogString(), config)
- }
- for key, value := range p.unitsMode {
- format += "\nUnitMode[%-v]: %-v"
- args = append(args, key.LogString(), value.LogString())
- }
- format += " ]>"
- return fmt.Sprintf(format, args...)
- }
-
- func applyEveryoneRepoPermission(user *user_model.User, perm *Permission) {
- if user != nil && user.ID > 0 {
- for _, u := range perm.units {
- if perm.unitsMode == nil {
- perm.unitsMode = make(map[unit.Type]perm_model.AccessMode)
- }
- if u.EveryoneAccessMode >= perm_model.AccessModeRead && u.EveryoneAccessMode > perm.unitsMode[u.Type] {
- perm.unitsMode[u.Type] = u.EveryoneAccessMode
- }
- }
- }
- }
-
- // GetUserRepoPermission returns the user permissions to the repository
- func GetUserRepoPermission(ctx context.Context, repo *repo_model.Repository, user *user_model.User) (perm Permission, err error) {
- defer func() {
- if err == nil {
- applyEveryoneRepoPermission(user, &perm)
- }
- if log.IsTrace() {
- log.Trace("Permission Loaded for user %-v in repo %-v, permissions: %-+v", user, repo, perm)
- }
- }()
-
- if err = repo.LoadUnits(ctx); err != nil {
- return perm, err
- }
- perm.units = repo.Units
-
- // anonymous user visit private repo.
- // TODO: anonymous user visit public unit of private repo???
- if user == nil && repo.IsPrivate {
- perm.AccessMode = perm_model.AccessModeNone
- return perm, nil
- }
-
- var isCollaborator bool
- if user != nil {
- isCollaborator, err = repo_model.IsCollaborator(ctx, repo.ID, user.ID)
- if err != nil {
- return perm, err
- }
- }
-
- if err = repo.LoadOwner(ctx); err != nil {
- return perm, err
- }
-
- // Prevent strangers from checking out public repo of private organization/users
- // Allow user if they are collaborator of a repo within a private user or a private organization but not a member of the organization itself
- if !organization.HasOrgOrUserVisible(ctx, repo.Owner, user) && !isCollaborator {
- perm.AccessMode = perm_model.AccessModeNone
- return perm, nil
- }
-
- // anonymous visit public repo
- if user == nil {
- perm.AccessMode = perm_model.AccessModeRead
- return perm, nil
- }
-
- // Admin or the owner has super access to the repository
- if user.IsAdmin || user.ID == repo.OwnerID {
- perm.AccessMode = perm_model.AccessModeOwner
- return perm, nil
- }
-
- // plain user
- perm.AccessMode, err = accessLevel(ctx, user, repo)
- if err != nil {
- return perm, err
- }
-
- if !repo.Owner.IsOrganization() {
- return perm, nil
- }
-
- perm.unitsMode = make(map[unit.Type]perm_model.AccessMode)
-
- // Collaborators on organization
- if isCollaborator {
- for _, u := range repo.Units {
- perm.unitsMode[u.Type] = perm.AccessMode
- }
- }
-
- // get units mode from teams
- teams, err := organization.GetUserRepoTeams(ctx, repo.OwnerID, user.ID, repo.ID)
- if err != nil {
- return perm, err
- }
-
- // if user in an owner team
- for _, team := range teams {
- if team.AccessMode >= perm_model.AccessModeAdmin {
- perm.AccessMode = perm_model.AccessModeOwner
- perm.unitsMode = nil
- return perm, nil
- }
- }
-
- for _, u := range repo.Units {
- var found bool
- for _, team := range teams {
- if teamMode, exist := team.UnitAccessModeEx(ctx, u.Type); exist {
- perm.unitsMode[u.Type] = max(perm.unitsMode[u.Type], teamMode)
- found = true
- }
- }
-
- // for a public repo on an organization, a non-restricted user has read permission on non-team defined units.
- if !found && !repo.IsPrivate && !user.IsRestricted {
- if _, ok := perm.unitsMode[u.Type]; !ok {
- perm.unitsMode[u.Type] = perm_model.AccessModeRead
- }
- }
- }
-
- // remove no permission units
- perm.units = make([]*repo_model.RepoUnit, 0, len(repo.Units))
- for t := range perm.unitsMode {
- for _, u := range repo.Units {
- if u.Type == t {
- perm.units = append(perm.units, u)
- }
- }
- }
-
- return perm, err
- }
-
- // IsUserRealRepoAdmin check if this user is real repo admin
- func IsUserRealRepoAdmin(ctx context.Context, repo *repo_model.Repository, user *user_model.User) (bool, error) {
- if repo.OwnerID == user.ID {
- return true, nil
- }
-
- if err := repo.LoadOwner(ctx); err != nil {
- return false, err
- }
-
- accessMode, err := accessLevel(ctx, user, repo)
- if err != nil {
- return false, err
- }
-
- return accessMode >= perm_model.AccessModeAdmin, nil
- }
-
- // IsUserRepoAdmin return true if user has admin right of a repo
- func IsUserRepoAdmin(ctx context.Context, repo *repo_model.Repository, user *user_model.User) (bool, error) {
- if user == nil || repo == nil {
- return false, nil
- }
- if user.IsAdmin {
- return true, nil
- }
-
- mode, err := accessLevel(ctx, user, repo)
- if err != nil {
- return false, err
- }
- if mode >= perm_model.AccessModeAdmin {
- return true, nil
- }
-
- teams, err := organization.GetUserRepoTeams(ctx, repo.OwnerID, user.ID, repo.ID)
- if err != nil {
- return false, err
- }
-
- for _, team := range teams {
- if team.AccessMode >= perm_model.AccessModeAdmin {
- return true, nil
- }
- }
- return false, nil
- }
-
- // AccessLevel returns the Access a user has to a repository. Will return NoneAccess if the
- // user does not have access.
- func AccessLevel(ctx context.Context, user *user_model.User, repo *repo_model.Repository) (perm_model.AccessMode, error) { //nolint
- return AccessLevelUnit(ctx, user, repo, unit.TypeCode)
- }
-
- // AccessLevelUnit returns the Access a user has to a repository's. Will return NoneAccess if the
- // user does not have access.
- func AccessLevelUnit(ctx context.Context, user *user_model.User, repo *repo_model.Repository, unitType unit.Type) (perm_model.AccessMode, error) { //nolint
- perm, err := GetUserRepoPermission(ctx, repo, user)
- if err != nil {
- return perm_model.AccessModeNone, err
- }
- return perm.UnitAccessMode(unitType), nil
- }
-
- // HasAccessUnit returns true if user has testMode to the unit of the repository
- func HasAccessUnit(ctx context.Context, user *user_model.User, repo *repo_model.Repository, unitType unit.Type, testMode perm_model.AccessMode) (bool, error) {
- mode, err := AccessLevelUnit(ctx, user, repo, unitType)
- return testMode <= mode, err
- }
-
- // CanBeAssigned return true if user can be assigned to issue or pull requests in repo
- // Currently any write access (code, issues or pr's) is assignable, to match assignee list in user interface.
- func CanBeAssigned(ctx context.Context, user *user_model.User, repo *repo_model.Repository, _ bool) (bool, error) {
- if user.IsOrganization() {
- return false, fmt.Errorf("organization can't be added as assignee [user_id: %d, repo_id: %d]", user.ID, repo.ID)
- }
- perm, err := GetUserRepoPermission(ctx, repo, user)
- if err != nil {
- return false, err
- }
- return perm.CanAccessAny(perm_model.AccessModeWrite, unit.AllRepoUnitTypes...) ||
- perm.CanAccessAny(perm_model.AccessModeRead, unit.TypePullRequests), nil
- }
-
- // HasAccess returns true if user has access to repo
- func HasAccess(ctx context.Context, userID int64, repo *repo_model.Repository) (bool, error) {
- var user *user_model.User
- var err error
- if userID > 0 {
- user, err = user_model.GetUserByID(ctx, userID)
- if err != nil {
- return false, err
- }
- }
- perm, err := GetUserRepoPermission(ctx, repo, user)
- if err != nil {
- return false, err
- }
- return perm.HasAccess(), nil
- }
-
- // getUsersWithAccessMode returns users that have at least given access mode to the repository.
- func getUsersWithAccessMode(ctx context.Context, repo *repo_model.Repository, mode perm_model.AccessMode) (_ []*user_model.User, err error) {
- if err = repo.LoadOwner(ctx); err != nil {
- return nil, err
- }
-
- e := db.GetEngine(ctx)
- accesses := make([]*Access, 0, 10)
- if err = e.Where("repo_id = ? AND mode >= ?", repo.ID, mode).Find(&accesses); err != nil {
- return nil, err
- }
-
- // Leave a seat for owner itself to append later, but if owner is an organization
- // and just waste 1 unit is cheaper than re-allocate memory once.
- users := make([]*user_model.User, 0, len(accesses)+1)
- if len(accesses) > 0 {
- userIDs := make([]int64, len(accesses))
- for i := 0; i < len(accesses); i++ {
- userIDs[i] = accesses[i].UserID
- }
-
- if err = e.In("id", userIDs).Find(&users); err != nil {
- return nil, err
- }
- }
- if !repo.Owner.IsOrganization() {
- users = append(users, repo.Owner)
- }
-
- return users, nil
- }
-
- // GetRepoReaders returns all users that have explicit read access or higher to the repository.
- func GetRepoReaders(ctx context.Context, repo *repo_model.Repository) (_ []*user_model.User, err error) {
- return getUsersWithAccessMode(ctx, repo, perm_model.AccessModeRead)
- }
-
- // GetRepoWriters returns all users that have write access to the repository.
- func GetRepoWriters(ctx context.Context, repo *repo_model.Repository) (_ []*user_model.User, err error) {
- return getUsersWithAccessMode(ctx, repo, perm_model.AccessModeWrite)
- }
-
- // IsRepoReader returns true if user has explicit read access or higher to the repository.
- func IsRepoReader(ctx context.Context, repo *repo_model.Repository, userID int64) (bool, error) {
- if repo.OwnerID == userID {
- return true, nil
- }
- return db.GetEngine(ctx).Where("repo_id = ? AND user_id = ? AND mode >= ?", repo.ID, userID, perm_model.AccessModeRead).Get(&Access{})
- }
-
- // CheckRepoUnitUser check whether user could visit the unit of this repository
- func CheckRepoUnitUser(ctx context.Context, repo *repo_model.Repository, user *user_model.User, unitType unit.Type) bool {
- if user != nil && user.IsAdmin {
- return true
- }
- perm, err := GetUserRepoPermission(ctx, repo, user)
- if err != nil {
- log.Error("GetUserRepoPermission: %w", err)
- return false
- }
-
- return perm.CanRead(unitType)
- }
|