mirror of
https://github.com/go-gitea/gitea
synced 2024-12-23 00:54:28 +00:00
48d4580dd5
Follow #30495 "HasAccess" behavior wasn't clear, to make it clear: * Use a new name `HasAnyUnitAccess`, it will be easier to review related code and permission problems. * Separate everyone access mode to a separate field, then all calls to HasAccess are reverted to old behavior before #30495. * Add new tests. --------- Co-authored-by: Giteabot <teabot@gitea.io>
474 lines
15 KiB
Go
474 lines
15 KiB
Go
// 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
|
|
|
|
everyoneAccessMode 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
|
|
}
|
|
|
|
// HasAnyUnitAccess returns true if the user might have at least one access mode to any unit of this repository.
|
|
// It doesn't count the "everyone access mode".
|
|
func (p *Permission) HasAnyUnitAccess() bool {
|
|
for _, v := range p.unitsMode {
|
|
if v >= perm_model.AccessModeRead {
|
|
return true
|
|
}
|
|
}
|
|
return p.AccessMode >= perm_model.AccessModeRead
|
|
}
|
|
|
|
func (p *Permission) HasAnyUnitAccessOrEveryoneAccess() bool {
|
|
for _, v := range p.everyoneAccessMode {
|
|
if v >= perm_model.AccessModeRead {
|
|
return true
|
|
}
|
|
}
|
|
return p.HasAnyUnitAccess()
|
|
}
|
|
|
|
// 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
|
|
// It also considers "everyone access mode"
|
|
func (p *Permission) UnitAccessMode(unitType unit.Type) perm_model.AccessMode {
|
|
// 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
|
|
unitDefaultAccessMode := max(p.AccessMode, p.everyoneAccessMode[unitType])
|
|
hasUnit := slices.ContainsFunc(p.units, func(u *repo_model.RepoUnit) bool { return u.Type == unitType })
|
|
return util.Iif(hasUnit, unitDefaultAccessMode, 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 {
|
|
return
|
|
}
|
|
for _, u := range perm.units {
|
|
if u.EveryoneAccessMode >= perm_model.AccessModeRead && u.EveryoneAccessMode > perm.everyoneAccessMode[u.Type] {
|
|
if perm.everyoneAccessMode == nil {
|
|
perm.everyoneAccessMode = make(map[unit.Type]perm_model.AccessMode)
|
|
}
|
|
perm.everyoneAccessMode[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
|
|
}
|
|
|
|
// HasAnyUnitAccess see the comment of "perm.HasAnyUnitAccess"
|
|
func HasAnyUnitAccess(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.HasAnyUnitAccess(), 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)
|
|
}
|